clean

### Describe Your Changes

Some messages were written to `stdout` using `fmt.Printf` and
`fmt.Println`, while the other messages like import statistics were
written to `stderr` through the `log` package.

This led to ordering problems where the `Import finished!` +
`VictoriaMetrics importer stats` messages, which expected to be the last
messages, appeared before `Continue import process with filter`
messages, creating confusing output for users.

```
2025/08/20 13:07:26 Import finished!
2025/08/20 13:07:26 VictoriaMetrics importer stats:
  time spent while importing: 20h49m10.8497184s;
  total bytes: 277.1 GB;
  bytes/s: 3.7 MB;
  requests: 7978614;
  requests retries: 0;
2025/08/20 13:07:26 Total time: 20h49m10.851006088s
Continue import process with filter
        filter: match[]={__name__!=""}
        start: 2025-08-08T00:00:00Z
        end: 2025-08-15T00:00:00Z:
Continue import process with filter
        filter: match[]={__name__!=""}
        start: 2025-08-15T00:00:00Z
        end: 2025-08-19T16:18:15Z:
```


### Checklist

The following checks are **mandatory**:

- [x] My change adheres to [VictoriaMetrics contributing
guidelines](https://docs.victoriametrics.com/victoriametrics/contributing/#pull-request-checklist).
- [x] My change adheres to [VictoriaMetrics development
goals](https://docs.victoriametrics.com/victoriametrics/goals/).

.github/copilot-instructions.md

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

.github/workflows/build.yml

@@ -7,16 +7,20 @@ on:
       - master
     paths:
       - '**.go'
-      - '**/Dockerfile*' # The trailing * is for app/vmui/Dockerfile-*.
+      - '**/Dockerfile'
       - '**/Makefile'
+      - '!app/vmui/**'
+      - '.github/workflows/build.yml'
   pull_request:
     branches:
       - cluster
       - master
     paths:
       - '**.go'
-      - '**/Dockerfile*' # The trailing * is for app/vmui/Dockerfile-*.
+      - '**/Dockerfile'
       - '**/Makefile'
+      - '!app/vmui/**'
+      - '.github/workflows/build.yml'
 
 permissions:
   contents: read
@@ -27,28 +31,48 @@ concurrency:
 
 jobs:
   build:
-    name: Build
+    name: ${{ matrix.os }}-${{ matrix.arch }}
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - os: linux
+            arch: 386
+          - os: linux
+            arch: amd64
+          - os: linux
+            arch: arm64
+          - os: linux
+            arch: arm
+          - os: linux
+            arch: ppc64le
+          - os: darwin
+            arch: amd64
+          - os: darwin
+            arch: arm64
+          - os: freebsd
+            arch: amd64
+          - os: openbsd
+            arch: amd64
+          - os: windows
+            arch: amd64
     steps:
       - name: Code checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Setup Go
         id: go
         uses: actions/setup-go@v5
         with:
+          cache-dependency-path: |
+            go.sum
+            Makefile
+            app/**/Makefile
           go-version: stable
-          cache: false
 
-      - name: Cache Go artifacts
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-crossbuild-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-crossbuild-
+      - name: Build victoria-metrics for ${{ matrix.os }}-${{ matrix.arch }}
+        run: make victoria-metrics-${{ matrix.os }}-${{ matrix.arch }}
 
-      - name: Run crossbuild
-        run: make crossbuild
+      - name: Build vmutils for ${{ matrix.os }}-${{ matrix.arch }}
+        run: make vmutils-${{ matrix.os }}-${{ matrix.arch }}

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

@@ -29,7 +29,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Set up Go
         id: go

.github/workflows/codeql-analysis-js-ts.yml

@@ -1,46 +0,0 @@
-name: 'CodeQL JS/TS'
-
-on:
-  push:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.js'
-      - '**.ts'
-      - '**.tsx'
-  pull_request:
-    branches:
-      - cluster
-      - master
-    paths:
-      - '**.js'
-      - '**.ts'
-      - '**.tsx'
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
-  cancel-in-progress: true
-
-jobs:
-  analyze:
-    name: Analyze
-    runs-on: ubuntu-latest
-    permissions:
-      actions: read
-      contents: read
-      security-events: write
-
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-
-      - name: Initialize CodeQL
-        uses: github/codeql-action/init@v3
-        with:
-          languages: javascript-typescript
-
-      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v3
-        with:
-          category: 'language:js/ts'

.github/workflows/docs.yaml

@@ -16,12 +16,12 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           path: __vm
 
       - name: Checkout private code
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
         with:
           repository: VictoriaMetrics/vmdocs
           token: ${{ secrets.VM_BOT_GH_TOKEN }}

.github/workflows/test.yml

@@ -1,4 +1,4 @@
-name: main
+name: test
 
 on:
   push:
@@ -7,12 +7,16 @@ on:
       - master
     paths:
       - '**.go'
+      - 'go.*'
+      - '.github/workflows/main.yml'
   pull_request:
     branches:
       - cluster
       - master
     paths:
       - '**.go'
+      - 'go.*'
+      - '.github/workflows/main.yml'
 
 permissions:
   contents: read
@@ -21,39 +25,41 @@ concurrency:
   cancel-in-progress: true
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
 
+
 jobs:
   lint:
     name: lint
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Setup Go
         id: go
         uses: actions/setup-go@v5
         with:
-          cache: false
+          cache-dependency-path: |
+            go.sum
+            Makefile
+            app/**/Makefile
           go-version: stable
 
-      - name: Cache Go artifacts
+
+      - name: Cache golangci-lint
         uses: actions/cache@v4
         with:
           path: |
-            ~/.cache/go-build
+            ~/.cache/golangci-lint
             ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-check-all-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-check-all-
+          key: golangci-lint-${{ runner.os }}-${{ hashFiles('.golangci.yml') }}
 
       - name: Run check-all
         run: |
           make check-all
           git diff --exit-code
 
-  test:
-    name: test
-    needs: lint
+  unit:
+    name: unit
     runs-on: ubuntu-latest
 
     strategy:
@@ -65,25 +71,18 @@ jobs:
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Setup Go
         id: go
         uses: actions/setup-go@v5
         with:
-          cache: false
+          cache-dependency-path: |
+            go.sum
+            Makefile
+            app/**/Makefile
           go-version: stable
 
-      - name: Cache Go artifacts
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-
-
       - name: Run tests
         run: GOGC=10 make ${{ matrix.scenario}}
 
@@ -92,31 +91,23 @@ jobs:
         with:
           files: ./coverage.txt
 
-  integration-test:
-    name: integration-test
-    needs: [lint, test]
+  integration:
+    name: integration
     runs-on: ubuntu-latest
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Setup Go
         id: go
         uses: actions/setup-go@v5
         with:
-          cache: false
+          cache-dependency-path: |
+            go.sum
+            Makefile
+            app/**/Makefile
           go-version: stable
 
-      - name: Cache Go artifacts
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-
-
       - name: Run integration tests
         run: make integration-test

.github/workflows/vmui.yml

@@ -0,0 +1,82 @@
+name: vmui
+
+on:
+  push:
+    branches:
+      - cluster
+      - master
+    paths:
+      - 'app/vmui/packages/vmui/**'
+      - '.github/workflows/vmui.yml'
+  pull_request:
+    branches:
+      - cluster
+      - master
+    paths:
+      - 'app/vmui/packages/vmui/**'
+      - '.github/workflows/vmui.yml'
+
+permissions:
+  contents: read
+  packages: read
+  pull-requests: read
+  checks: write
+
+concurrency:
+  cancel-in-progress: true
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+
+jobs:
+  vmui-checks:
+    name: VMUI Checks (lint, test, typecheck)
+    runs-on: ubuntu-latest
+    steps:
+      - name: Code checkout
+        uses: actions/checkout@v5
+
+      - name: Setup Node
+        uses: actions/setup-node@v4
+        with:
+          node-version: '24.x'
+
+      - name: Cache node-modules
+        uses: actions/cache@v4
+        with:
+          path: |
+            app/vmui/packages/vmui/node_modules
+          key: vmui-artifacts-${{ runner.os }}-${{ hashFiles('package-lock.json') }}
+          restore-keys: vmui-artifacts-${{ runner.os }}-
+
+      - name: Run lint
+        id: lint
+        run: make vmui-lint
+        continue-on-error: true
+
+      - name: Run tests
+        id: test
+        run: make vmui-test
+        continue-on-error: true
+
+      - name: Run typecheck
+        id: typecheck
+        run: make vmui-typecheck
+        continue-on-error: true
+
+      - name: Annotate Code Linting Results
+        uses: ataylorme/eslint-annotate-action@v3
+        with:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          report-json: app/vmui/packages/vmui/vmui-lint-report.json
+
+      - name: Check overall status
+        run: |
+          echo "Lint status: ${{ steps.lint.outcome }}"
+          echo "Test status: ${{ steps.test.outcome }}"
+          echo "Typecheck status: ${{ steps.typecheck.outcome }}"
+
+          if [[ "${{ steps.lint.outcome }}" == "failure" || "${{ steps.test.outcome }}" == "failure" || "${{ steps.typecheck.outcome }}" == "failure" ]]; then
+            echo "One or more checks failed"
+            exit 1
+          else
+            echo "All checks passed"
+          fi

.golangci.yml

@@ -1,22 +1,29 @@
-run:
-  timeout: 2m
-
+version: "2"
 linters:
-  enable:
-  - revive
-
-issues:
-  exclude-rules:
-  - linters:
-    - staticcheck
-    text: "SA(4003|1019|5011):"
-  include:
-  - EXC0012
-  - EXC0014
-
-linters-settings:
-  errcheck:
-    exclude-functions:
-      - "fmt.Fprintf"
-      - "fmt.Fprint"
-      - "(net/http.ResponseWriter).Write"
+  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:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

Makefile

@@ -11,11 +11,14 @@ ifeq ($(PKG_TAG),)
 PKG_TAG := $(BUILDINFO_TAG)
 endif
 
-EXTRA_DOCKER_TAG_SUFFIX ?= EXTRA_DOCKER_TAG_SUFFIX
+EXTRA_DOCKER_TAG_SUFFIX ?=
+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.4.0
+
 .PHONY: $(MAKECMDGOALS)
 
 include app/*/Makefile
@@ -24,11 +27,10 @@ include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
 include package/release/Makefile
+include benchmarks/Makefile
 
 all: \
 	victoria-metrics-prod \
-	victoria-logs-prod \
-	vlogscli-prod \
 	vmagent-prod \
 	vmalert-prod \
 	vmalert-tool-prod \
@@ -52,8 +54,6 @@ publish: \
 
 package: \
 	package-victoria-metrics \
-	package-victoria-logs \
-	package-vlogscli \
 	package-vmagent \
 	package-vmalert \
 	package-vmalert-tool \
@@ -170,9 +170,11 @@ vmutils-windows-amd64: \
 	vmrestore-windows-amd64 \
 	vmctl-windows-amd64
 
+# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 crossbuild:
 	$(MAKE_PARALLEL) victoria-metrics-crossbuild vmutils-crossbuild
 
+# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 victoria-metrics-crossbuild: \
 	victoria-metrics-linux-386 \
 	victoria-metrics-linux-amd64 \
@@ -185,6 +187,7 @@ victoria-metrics-crossbuild: \
 	victoria-metrics-openbsd-amd64 \
 	victoria-metrics-windows-amd64
 
+# When adding a new crossbuild target, please also add it to the .github/workflows/build.yml
 vmutils-crossbuild: \
 	vmutils-linux-386 \
 	vmutils-linux-amd64 \
@@ -237,10 +240,6 @@ publish-latest:
 	PKG_TAG=$(TAG)-enterprise APP_NAME=vmgateway $(MAKE) publish-via-docker-latest
 	PKG_TAG=$(TAG)-enterprise APP_NAME=vmbackupmanager $(MAKE) publish-via-docker-latest
 
-publish-victoria-logs-latest:
-	PKG_TAG=$(TAG) APP_NAME=victoria-logs $(MAKE) publish-via-docker-latest
-	PKG_TAG=$(TAG) APP_NAME=vlogscli $(MAKE) publish-via-docker-latest
-
 publish-release:
 	rm -rf bin/*
 	git checkout $(TAG) && $(MAKE) release && $(MAKE) publish && \
@@ -310,128 +309,6 @@ release-victoria-metrics-windows-goarch: victoria-metrics-windows-$(GOARCH)-prod
 	cd bin && rm -rf \
 		victoria-metrics-windows-$(GOARCH)-prod.exe
 
-release-victoria-logs-bundle: \
-	release-victoria-logs \
-	release-vlogscli
-
-publish-victoria-logs-bundle: \
-	publish-victoria-logs \
-	publish-vlogscli
-
-release-victoria-logs:
-	$(MAKE_PARALLEL) release-victoria-logs-linux-386 \
-		release-victoria-logs-linux-amd64 \
-		release-victoria-logs-linux-arm \
-		release-victoria-logs-linux-arm64 \
-		release-victoria-logs-darwin-amd64 \
-		release-victoria-logs-darwin-arm64 \
-		release-victoria-logs-freebsd-amd64 \
-		release-victoria-logs-openbsd-amd64 \
-		release-victoria-logs-windows-amd64
-
-release-victoria-logs-linux-386:
-	GOOS=linux GOARCH=386 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-linux-amd64:
-	GOOS=linux GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-linux-arm:
-	GOOS=linux GOARCH=arm $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-linux-arm64:
-	GOOS=linux GOARCH=arm64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-darwin-amd64:
-	GOOS=darwin GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-darwin-arm64:
-	GOOS=darwin GOARCH=arm64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-freebsd-amd64:
-	GOOS=freebsd GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-openbsd-amd64:
-	GOOS=openbsd GOARCH=amd64 $(MAKE) release-victoria-logs-goos-goarch
-
-release-victoria-logs-windows-amd64:
-	GOARCH=amd64 $(MAKE) release-victoria-logs-windows-goarch
-
-release-victoria-logs-goos-goarch: victoria-logs-$(GOOS)-$(GOARCH)-prod
-	cd bin && \
-		tar $(TAR_OWNERSHIP) --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
-			victoria-logs-$(GOOS)-$(GOARCH)-prod \
-		&& sha256sum victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
-			victoria-logs-$(GOOS)-$(GOARCH)-prod \
-			| sed s/-$(GOOS)-$(GOARCH)-prod/-prod/ > victoria-logs-$(GOOS)-$(GOARCH)-$(PKG_TAG)_checksums.txt
-	cd bin && rm -rf victoria-logs-$(GOOS)-$(GOARCH)-prod
-
-release-victoria-logs-windows-goarch: victoria-logs-windows-$(GOARCH)-prod
-	cd bin && \
-		zip victoria-logs-windows-$(GOARCH)-$(PKG_TAG).zip \
-			victoria-logs-windows-$(GOARCH)-prod.exe \
-		&& sha256sum victoria-logs-windows-$(GOARCH)-$(PKG_TAG).zip \
-			victoria-logs-windows-$(GOARCH)-prod.exe \
-			> victoria-logs-windows-$(GOARCH)-$(PKG_TAG)_checksums.txt
-	cd bin && rm -rf \
-		victoria-logs-windows-$(GOARCH)-prod.exe
-
-release-vlogscli:
-	$(MAKE_PARALLEL) release-vlogscli-linux-386 \
-		release-vlogscli-linux-amd64 \
-		release-vlogscli-linux-arm \
-		release-vlogscli-linux-arm64 \
-		release-vlogscli-darwin-amd64 \
-		release-vlogscli-darwin-arm64 \
-		release-vlogscli-freebsd-amd64 \
-		release-vlogscli-openbsd-amd64 \
-		release-vlogscli-windows-amd64
-
-release-vlogscli-linux-386:
-	GOOS=linux GOARCH=386 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-linux-amd64:
-	GOOS=linux GOARCH=amd64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-linux-arm:
-	GOOS=linux GOARCH=arm $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-linux-arm64:
-	GOOS=linux GOARCH=arm64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-darwin-amd64:
-	GOOS=darwin GOARCH=amd64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-darwin-arm64:
-	GOOS=darwin GOARCH=arm64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-freebsd-amd64:
-	GOOS=freebsd GOARCH=amd64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-openbsd-amd64:
-	GOOS=openbsd GOARCH=amd64 $(MAKE) release-vlogscli-goos-goarch
-
-release-vlogscli-windows-amd64:
-	GOARCH=amd64 $(MAKE) release-vlogscli-windows-goarch
-
-release-vlogscli-goos-goarch: vlogscli-$(GOOS)-$(GOARCH)-prod
-	cd bin && \
-		tar $(TAR_OWNERSHIP) --transform="flags=r;s|-$(GOOS)-$(GOARCH)||" -czf vlogscli-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
-			vlogscli-$(GOOS)-$(GOARCH)-prod \
-		&& sha256sum vlogscli-$(GOOS)-$(GOARCH)-$(PKG_TAG).tar.gz \
-			vlogscli-$(GOOS)-$(GOARCH)-prod \
-			| sed s/-$(GOOS)-$(GOARCH)-prod/-prod/ > vlogscli-$(GOOS)-$(GOARCH)-$(PKG_TAG)_checksums.txt
-	cd bin && rm -rf vlogscli-$(GOOS)-$(GOARCH)-prod
-
-release-vlogscli-windows-goarch: vlogscli-windows-$(GOARCH)-prod
-	cd bin && \
-		zip vlogscli-windows-$(GOARCH)-$(PKG_TAG).zip \
-			vlogscli-windows-$(GOARCH)-prod.exe \
-		&& sha256sum vlogscli-windows-$(GOARCH)-$(PKG_TAG).zip \
-			vlogscli-windows-$(GOARCH)-prod.exe \
-			> vlogscli-windows-$(GOARCH)-$(PKG_TAG)_checksums.txt
-	cd bin && rm -rf \
-		vlogscli-windows-$(GOARCH)-prod.exe
-
 release-vmutils: \
 	release-vmutils-linux-386 \
 	release-vmutils-linux-amd64 \
@@ -572,7 +449,11 @@ test-full:
 test-full-386:
 	GOEXPERIMENT=synctest GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
-integration-test: victoria-metrics vmagent vmalert vmauth vmctl vmbackup vmrestore
+integration-test:
+	$(MAKE) apptest
+
+apptest:
+	$(MAKE) victoria-metrics vmagent vmalert vmauth vmctl vmbackup vmrestore
 	go test ./apptest/... -skip="^TestCluster.*"
 
 benchmark:
@@ -590,16 +471,16 @@ vendor-update:
 	go mod vendor
 
 app-local:
-	CGO_ENABLED=1 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=1 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-pure:
-	CGO_ENABLED=0 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-pure$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=0 go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-pure$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-goos-goarch:
-	CGO_ENABLED=$(CGO_ENABLED) GOOS=$(GOOS) GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-$(GOOS)-$(GOARCH)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=$(CGO_ENABLED) GOOS=$(GOOS) GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-$(GOOS)-$(GOARCH)$(RACE) $(PKG_PREFIX)/app/$(APP_NAME)
 
 app-local-windows-goarch:
-	CGO_ENABLED=0 GOOS=windows GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -o bin/$(APP_NAME)-windows-$(GOARCH)$(RACE).exe $(PKG_PREFIX)/app/$(APP_NAME)
+	CGO_ENABLED=0 GOOS=windows GOARCH=$(GOARCH) go build $(RACE) -ldflags "$(GO_BUILDINFO)" -tags "$(EXTRA_GO_BUILD_TAGS)" -o bin/$(APP_NAME)-windows-$(GOARCH)$(RACE).exe $(PKG_PREFIX)/app/$(APP_NAME)
 
 quicktemplate-gen: install-qtc
 	qtc
@@ -612,7 +493,7 @@ golangci-lint: install-golangci-lint
 	GOEXPERIMENT=synctest golangci-lint run
 
 install-golangci-lint:
-	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.7
+	which golangci-lint && (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`

app/victoria-metrics/main.go

@@ -17,7 +17,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
@@ -125,8 +124,6 @@ func main() {
 	vmstorage.Stop()
 	vmselect.Stop()
 
-	fs.MustStopDirRemover()
-
 	logger.Infof("the VictoriaMetrics has been stopped in %.3f seconds", time.Since(startTime).Seconds())
 }
 

app/victoria-metrics/self_scraper.go

@@ -10,7 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
@@ -49,7 +49,7 @@ func selfScraper(scrapeInterval time.Duration) {
 	var bb bytesutil.ByteBuffer
 	var rows prometheus.Rows
 	var mrs []storage.MetricRow
-	var labels []prompbmarshal.Label
+	var labels []prompb.Label
 	t := time.NewTicker(scrapeInterval)
 	f := func(currentTime time.Time, sendStaleMarkers bool) {
 		currentTimestamp := currentTime.UnixNano() / 1e6
@@ -57,7 +57,8 @@ func selfScraper(scrapeInterval time.Duration) {
 		appmetrics.WritePrometheusMetrics(&bb)
 		s := bytesutil.ToUnsafeString(bb.B)
 		rows.Reset()
-		rows.Unmarshal(s)
+		// VictoriaMetrics components don't expose metadata yet, only need to parse samples
+		rows.UnmarshalWithErrLogger(s, nil)
 		mrs = mrs[:0]
 		for i := range rows.Rows {
 			r := &rows.Rows[i]
@@ -104,11 +105,11 @@ func selfScraper(scrapeInterval time.Duration) {
 	}
 }
 
-func addLabel(dst []prompbmarshal.Label, key, value string) []prompbmarshal.Label {
+func addLabel(dst []prompb.Label, key, value string) []prompb.Label {
 	if len(dst) < cap(dst) {
 		dst = dst[:len(dst)+1]
 	} else {
-		dst = append(dst, prompbmarshal.Label{})
+		dst = append(dst, prompb.Label{})
 	}
 	lb := &dst[len(dst)-1]
 	lb.Name = key

app/vmagent/common/push_ctx.go

@@ -3,7 +3,7 @@ package common
 import (
 	"sync"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -12,13 +12,13 @@ type PushCtx struct {
 	// WriteRequest contains the WriteRequest, which must be pushed later to remote storage.
 	//
 	// The actual labels and samples for the time series are stored in Labels and Samples fields.
-	WriteRequest prompbmarshal.WriteRequest
+	WriteRequest prompb.WriteRequest
 
 	// Labels contains flat list of all the labels used in WriteRequest.
-	Labels []prompbmarshal.Label
+	Labels []prompb.Label
 
 	// Samples contains flat list of all the samples used in WriteRequest.
-	Samples []prompbmarshal.Sample
+	Samples []prompb.Sample
 }
 
 // Reset resets ctx.

app/vmagent/csvimport/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -31,7 +31,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, rows []csvimport.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []csvimport.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -41,23 +41,23 @@ func insertRows(at *auth.Token, rows []csvimport.Row, extraLabels []prompbmarsha
 	for i := range rows {
 		r := &rows[i]
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: r.Metric,
 		})
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
 		labels = append(labels, extraLabels...)
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     r.Value,
 			Timestamp: r.Timestamp,
 		})
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[len(samples)-1:],
 		})

app/vmagent/datadogsketches/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
@@ -33,7 +33,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -45,12 +45,12 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 		ms := sketch.ToSummary()
 		for _, m := range ms {
 			labelsLen := len(labels)
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "__name__",
 				Value: m.Name,
 			})
 			for _, label := range m.Labels {
-				labels = append(labels, prompbmarshal.Label{
+				labels = append(labels, prompb.Label{
 					Name:  label.Name,
 					Value: label.Value,
 				})
@@ -60,7 +60,7 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 				if name == "host" {
 					name = "exported_host"
 				}
-				labels = append(labels, prompbmarshal.Label{
+				labels = append(labels, prompb.Label{
 					Name:  name,
 					Value: value,
 				})
@@ -68,13 +68,13 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 			labels = append(labels, extraLabels...)
 			samplesLen := len(samples)
 			for _, p := range m.Points {
-				samples = append(samples, prompbmarshal.Sample{
+				samples = append(samples, prompb.Sample{
 					Timestamp: p.Timestamp,
 					Value:     p.Value,
 				})
 			}
 			rowsTotal += len(m.Points)
-			tssDst = append(tssDst, prompbmarshal.TimeSeries{
+			tssDst = append(tssDst, prompb.TimeSeries{
 				Labels:  labels[labelsLen:],
 				Samples: samples[samplesLen:],
 			})

app/vmagent/datadogv1/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1/stream"
@@ -33,7 +33,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -45,18 +45,18 @@ func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompbm
 		ss := &series[i]
 		rowsTotal += len(ss.Points)
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: ss.Metric,
 		})
 		if ss.Host != "" {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "host",
 				Value: ss.Host,
 			})
 		}
 		if ss.Device != "" {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "device",
 				Value: ss.Device,
 			})
@@ -66,7 +66,7 @@ func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompbm
 			if name == "host" {
 				name = "exported_host"
 			}
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  name,
 				Value: value,
 			})
@@ -74,12 +74,12 @@ func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompbm
 		labels = append(labels, extraLabels...)
 		samplesLen := len(samples)
 		for _, pt := range ss.Points {
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Timestamp: pt.Timestamp(),
 				Value:     pt.Value(),
 			})
 		}
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})

app/vmagent/datadogv2/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2/stream"
@@ -36,7 +36,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -48,18 +48,18 @@ func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompbm
 		ss := &series[i]
 		rowsTotal += len(ss.Points)
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: ss.Metric,
 		})
 		for _, rs := range ss.Resources {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  rs.Type,
 				Value: rs.Name,
 			})
 		}
 		if ss.SourceTypeName != "" {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "source_type_name",
 				Value: ss.SourceTypeName,
 			})
@@ -69,7 +69,7 @@ func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompbm
 			if name == "host" {
 				name = "exported_host"
 			}
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  name,
 				Value: value,
 			})
@@ -77,12 +77,12 @@ func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompbm
 		labels = append(labels, extraLabels...)
 		samplesLen := len(samples)
 		for _, pt := range ss.Points {
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Timestamp: pt.Timestamp * 1000,
 				Value:     pt.Value,
 			})
 		}
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})

app/vmagent/graphite/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/graphite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/graphite/stream"
 	"github.com/VictoriaMetrics/metrics"
@@ -36,22 +36,22 @@ func insertRows(at *auth.Token, rows []parser.Row) error {
 	for i := range rows {
 		r := &rows[i]
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: r.Metric,
 		})
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     r.Value,
 			Timestamp: r.Timestamp,
 		})
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[len(samples)-1:],
 		})

app/vmagent/influx/request_handler.go

@@ -10,7 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx/stream"
@@ -60,7 +60,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prompb.Label) error {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)
 
@@ -80,13 +80,13 @@ func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prom
 			if tag.Key == *dbLabel {
 				hasDBKey = true
 			}
-			commonLabels = append(commonLabels, prompbmarshal.Label{
+			commonLabels = append(commonLabels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
 		if len(db) > 0 && !hasDBKey {
-			commonLabels = append(commonLabels, prompbmarshal.Label{
+			commonLabels = append(commonLabels, prompb.Label{
 				Name:  *dbLabel,
 				Value: db,
 			})
@@ -110,16 +110,16 @@ func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prom
 			}
 			metricGroup := bytesutil.ToUnsafeString(buf[bufLen:])
 			labelsLen := len(labels)
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "__name__",
 				Value: metricGroup,
 			})
 			labels = append(labels, commonLabels...)
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Timestamp: r.Timestamp,
 				Value:     f.Value,
 			})
-			tssDst = append(tssDst, prompbmarshal.TimeSeries{
+			tssDst = append(tssDst, prompb.TimeSeries{
 				Labels:  labels[labelsLen:],
 				Samples: samples[len(samples)-1:],
 			})
@@ -144,7 +144,7 @@ func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prom
 
 type pushCtx struct {
 	ctx            common.PushCtx
-	commonLabels   []prompbmarshal.Label
+	commonLabels   []prompb.Label
 	metricGroupBuf []byte
 	buf            []byte
 }

app/vmagent/main.go

@@ -206,7 +206,7 @@ func main() {
 func getOpenTSDBHTTPInsertHandler() func(req *http.Request) error {
 	if !remotewrite.MultitenancyEnabled() {
 		return func(req *http.Request) error {
-			path := strings.Replace(req.URL.Path, "//", "/", -1)
+			path := strings.ReplaceAll(req.URL.Path, "//", "/")
 			if path != "/api/put" {
 				return fmt.Errorf("unsupported path requested: %q; expecting '/api/put'", path)
 			}
@@ -214,7 +214,7 @@ func getOpenTSDBHTTPInsertHandler() func(req *http.Request) error {
 		}
 	}
 	return func(req *http.Request) error {
-		path := strings.Replace(req.URL.Path, "//", "/", -1)
+		path := strings.ReplaceAll(req.URL.Path, "//", "/")
 		at, err := getAuthTokenFromPath(path)
 		if err != nil {
 			return fmt.Errorf("cannot obtain auth token from path %q: %w", path, err)
@@ -259,7 +259,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 
-	path := strings.Replace(r.URL.Path, "//", "/", -1)
+	path := strings.ReplaceAll(r.URL.Path, "//", "/")
 	if strings.HasPrefix(path, "/prometheus/api/v1/import/prometheus") || strings.HasPrefix(path, "/api/v1/import/prometheus") {
 		prometheusimportRequests.Inc()
 		if err := prometheusimport.InsertHandler(nil, r); err != nil {

app/vmagent/native/request_handler.go

@@ -8,7 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
@@ -35,7 +35,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, block *stream.Block, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, block *stream.Block, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -53,13 +53,13 @@ func insertRows(at *auth.Token, block *stream.Block, extraLabels []prompbmarshal
 	samples := ctx.Samples[:0]
 	mn := &block.MetricName
 	labelsLen := len(labels)
-	labels = append(labels, prompbmarshal.Label{
+	labels = append(labels, prompb.Label{
 		Name:  "__name__",
 		Value: bytesutil.ToUnsafeString(mn.MetricGroup),
 	})
 	for j := range mn.Tags {
 		tag := &mn.Tags[j]
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  bytesutil.ToUnsafeString(tag.Key),
 			Value: bytesutil.ToUnsafeString(tag.Value),
 		})
@@ -72,12 +72,12 @@ func insertRows(at *auth.Token, block *stream.Block, extraLabels []prompbmarshal
 	}
 	samplesLen := len(samples)
 	for j, value := range values {
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     value,
 			Timestamp: timestamps[j],
 		})
 	}
-	tssDst = append(tssDst, prompbmarshal.TimeSeries{
+	tssDst = append(tssDst, prompb.TimeSeries{
 		Labels:  labels[labelsLen:],
 		Samples: samples[samplesLen:],
 	})

app/vmagent/newrelic/request_handler.go

@@ -9,7 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -34,7 +34,7 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, rows []newrelic.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []newrelic.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -49,22 +49,22 @@ func insertRows(at *auth.Token, rows []newrelic.Row, extraLabels []prompbmarshal
 		for j := range srcSamples {
 			s := &srcSamples[j]
 			labelsLen := len(labels)
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "__name__",
 				Value: bytesutil.ToUnsafeString(s.Name),
 			})
 			for k := range tags {
 				t := &tags[k]
-				labels = append(labels, prompbmarshal.Label{
+				labels = append(labels, prompb.Label{
 					Name:  bytesutil.ToUnsafeString(t.Key),
 					Value: bytesutil.ToUnsafeString(t.Value),
 				})
 			}
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Value:     s.Value,
 				Timestamp: r.Timestamp,
 			})
-			tssDst = append(tssDst, prompbmarshal.TimeSeries{
+			tssDst = append(tssDst, prompb.TimeSeries{
 				Labels:  labels[labelsLen:],
 				Samples: samples[len(samples)-1:],
 			})

app/vmagent/opentelemetry/request_handler.go

@@ -7,7 +7,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -16,9 +17,11 @@ import (
 )
 
 var (
-	rowsInserted       = metrics.NewCounter(`vmagent_rows_inserted_total{type="opentelemetry"}`)
-	rowsTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="opentelemetry"}`)
-	rowsPerInsert      = metrics.NewHistogram(`vmagent_rows_per_insert{type="opentelemetry"}`)
+	rowsInserted           = metrics.NewCounter(`vmagent_rows_inserted_total{type="opentelemetry"}`)
+	metadataInserted       = metrics.NewCounter(`vmagent_metadata_inserted_total{type="opentelemetry"}`)
+	rowsTenantInserted     = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="opentelemetry"}`)
+	metadataTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_metadata_total{type="opentelemetry"}`)
+	rowsPerInsert          = metrics.NewHistogram(`vmagent_rows_per_insert{type="opentelemetry"}`)
 )
 
 // InsertHandler processes opentelemetry metrics.
@@ -36,12 +39,12 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
 		}
 	}
-	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompbmarshal.TimeSeries) error {
-		return insertRows(at, tss, extraLabels)
+	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompb.TimeSeries, mms []prompb.MetricMetadata) error {
+		return insertRows(at, tss, mms, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, tss []prompbmarshal.TimeSeries, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, tss []prompb.TimeSeries, mms []prompb.MetricMetadata, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -57,20 +60,39 @@ func insertRows(at *auth.Token, tss []prompbmarshal.TimeSeries, extraLabels []pr
 		labels = append(labels, extraLabels...)
 		samplesLen := len(samples)
 		samples = append(samples, ts.Samples...)
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})
 	}
 	ctx.WriteRequest.Timeseries = tssDst
+
+	var metadataTotal int
+	if promscrape.IsMetadataEnabled() {
+		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)
+	}
+
 	ctx.Labels = labels
 	ctx.Samples = samples
 	if !remotewrite.TryPush(at, &ctx.WriteRequest) {
 		return remotewrite.ErrQueueFullHTTPRetry
 	}
 	rowsInserted.Add(rowsTotal)
+	metadataInserted.Add(metadataTotal)
 	if at != nil {
 		rowsTenantInserted.Get(at).Add(rowsTotal)
+		metadataTenantInserted.Get(at).Add(metadataTotal)
 	}
 	rowsPerInsert.Update(float64(rowsTotal))
 	return nil

app/vmagent/opentsdb/request_handler.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdb/stream"
 	"github.com/VictoriaMetrics/metrics"
@@ -33,22 +33,22 @@ func insertRows(rows []parser.Row) error {
 	for i := range rows {
 		r := &rows[i]
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: r.Metric,
 		})
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     r.Value,
 			Timestamp: r.Timestamp,
 		})
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[len(samples)-1:],
 		})

app/vmagent/opentsdbhttp/request_handler.go

@@ -6,7 +6,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -30,7 +30,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, rows []opentsdbhttp.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []opentsdbhttp.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -40,23 +40,23 @@ func insertRows(at *auth.Token, rows []opentsdbhttp.Row, extraLabels []prompbmar
 	for i := range rows {
 		r := &rows[i]
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: r.Metric,
 		})
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
 		labels = append(labels, extraLabels...)
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     r.Value,
 			Timestamp: r.Timestamp,
 		})
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[len(samples)-1:],
 		})

app/vmagent/prometheusimport/request_handler.go

@@ -7,7 +7,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -16,9 +17,12 @@ import (
 )
 
 var (
-	rowsInserted       = metrics.NewCounter(`vmagent_rows_inserted_total{type="prometheus"}`)
-	rowsTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="prometheus"}`)
-	rowsPerInsert      = metrics.NewHistogram(`vmagent_rows_per_insert{type="prometheus"}`)
+	rowsInserted           = metrics.NewCounter(`vmagent_rows_inserted_total{type="prometheus"}`)
+	metadataInserted       = metrics.NewCounter(`vmagent_metadata_inserted_total{type="prometheus"}`)
+	rowsTenantInserted     = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="prometheus"}`)
+	metadataTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_metadata_total{type="prometheus"}`)
+
+	rowsPerInsert = metrics.NewHistogram(`vmagent_rows_per_insert{type="prometheus"}`)
 )
 
 // InsertHandler processes `/api/v1/import/prometheus` request.
@@ -32,53 +36,74 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, defaultTimestamp, encoding, true, func(rows []prometheus.Row) error {
-		return insertRows(at, rows, extraLabels)
+	return stream.Parse(req.Body, defaultTimestamp, encoding, true, promscrape.IsMetadataEnabled(), func(rows []prometheus.Row, mms []prometheus.Metadata) error {
+		return insertRows(at, rows, mms, extraLabels)
 	}, func(s string) {
 		httpserver.LogError(req, s)
 	})
 }
 
-func insertRows(at *auth.Token, rows []prometheus.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []prometheus.Row, mms []prometheus.Metadata, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
 	tssDst := ctx.WriteRequest.Timeseries[:0]
+	mmsDst := ctx.WriteRequest.Metadata[:0]
 	labels := ctx.Labels[:0]
 	samples := ctx.Samples[:0]
 	for i := range rows {
 		r := &rows[i]
 		labelsLen := len(labels)
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: r.Metric,
 		})
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  tag.Key,
 				Value: tag.Value,
 			})
 		}
 		labels = append(labels, extraLabels...)
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Value:     r.Value,
 			Timestamp: r.Timestamp,
 		})
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			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{
+			MetricFamilyName: mm.Metric,
+			Help:             mm.Help,
+			Type:             mm.Type,
+			// there is no unit in Prometheus exposition formats
+
+			AccountID: accountID,
+			ProjectID: projectID,
+		})
+	}
 	ctx.WriteRequest.Timeseries = tssDst
+	ctx.WriteRequest.Metadata = mmsDst
 	ctx.Labels = labels
 	ctx.Samples = samples
 	if !remotewrite.TryPush(at, &ctx.WriteRequest) {
 		return remotewrite.ErrQueueFullHTTPRetry
 	}
 	rowsInserted.Add(len(rows))
+	metadataInserted.Add(len(mms))
 	if at != nil {
 		rowsTenantInserted.Get(at).Add(len(rows))
+		metadataTenantInserted.Get(at).Add(len(mms))
 	}
 	rowsPerInsert.Update(float64(len(rows)))
 	return nil

app/vmagent/prometheusimport/request_handler_test.go

@@ -55,6 +55,6 @@ func tearDown() {
 	srv.Close()
 	logger.ResetOutputForTest()
 	tmpDataDir := flag.Lookup("remoteWrite.tmpDataPath").Value.String()
-	fs.MustRemoveAll(tmpDataDir)
+	fs.MustRemoveDir(tmpDataDir)
 
 }

app/vmagent/promremotewrite/request_handler.go

@@ -7,7 +7,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
@@ -15,9 +15,11 @@ import (
 )
 
 var (
-	rowsInserted       = metrics.NewCounter(`vmagent_rows_inserted_total{type="promremotewrite"}`)
-	rowsTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="promremotewrite"}`)
-	rowsPerInsert      = metrics.NewHistogram(`vmagent_rows_per_insert{type="promremotewrite"}`)
+	rowsInserted           = metrics.NewCounter(`vmagent_rows_inserted_total{type="promremotewrite"}`)
+	metadataInserted       = metrics.NewCounter(`vmagent_metadata_inserted_total{type="promremotewrite"}`)
+	rowsTenantInserted     = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_rows_total{type="promremotewrite"}`)
+	metadataTenantInserted = tenantmetrics.NewCounterMap(`vmagent_tenant_inserted_metadata_total{type="promremotewrite"}`)
+	rowsPerInsert          = metrics.NewHistogram(`vmagent_rows_per_insert{type="promremotewrite"}`)
 )
 
 // InsertHandler processes remote write for prometheus.
@@ -27,17 +29,18 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 		return err
 	}
 	isVMRemoteWrite := req.Header.Get("Content-Encoding") == "zstd"
-	return stream.Parse(req.Body, isVMRemoteWrite, func(tss []prompb.TimeSeries) error {
-		return insertRows(at, tss, extraLabels)
+	return stream.Parse(req.Body, isVMRemoteWrite, func(tss []prompb.TimeSeries, mms []prompb.MetricMetadata) error {
+		return insertRows(at, tss, mms, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, mms []prompb.MetricMetadata, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
 	rowsTotal := 0
 	tssDst := ctx.WriteRequest.Timeseries[:0]
+	mmsDst := ctx.WriteRequest.Metadata[:0]
 	labels := ctx.Labels[:0]
 	samples := ctx.Samples[:0]
 	for i := range timeseries {
@@ -46,7 +49,7 @@ func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, extraLabels []pr
 		labelsLen := len(labels)
 		for i := range ts.Labels {
 			label := &ts.Labels[i]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  label.Name,
 				Value: label.Value,
 			})
@@ -55,17 +58,41 @@ func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, extraLabels []pr
 		samplesLen := len(samples)
 		for i := range ts.Samples {
 			sample := &ts.Samples[i]
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Value:     sample.Value,
 				Timestamp: sample.Timestamp,
 			})
 		}
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})
 	}
 	ctx.WriteRequest.Timeseries = tssDst
+
+	var metadataTotal int
+	if promscrape.IsMetadataEnabled() {
+		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{
+				MetricFamilyName: mm.MetricFamilyName,
+				Help:             mm.Help,
+				Type:             mm.Type,
+				Unit:             mm.Unit,
+
+				AccountID: accountID,
+				ProjectID: projectID,
+			})
+		}
+		ctx.WriteRequest.Metadata = mmsDst
+		metadataTotal = len(mms)
+	}
+
 	ctx.Labels = labels
 	ctx.Samples = samples
 	if !remotewrite.TryPush(at, &ctx.WriteRequest) {
@@ -74,7 +101,9 @@ func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, extraLabels []pr
 	rowsInserted.Add(rowsTotal)
 	if at != nil {
 		rowsTenantInserted.Get(at).Add(rowsTotal)
+		metadataTenantInserted.Get(at).Add(metadataTotal)
 	}
+	metadataInserted.Add(metadataTotal)
 	rowsPerInsert.Update(float64(rowsTotal))
 	return nil
 }

app/vmagent/remotewrite/client.go

@@ -448,7 +448,8 @@ again:
 	}
 
 	metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_requests_total{url=%q, status_code="%d"}`, c.sanitizedURL, statusCode)).Inc()
-	if statusCode == 409 {
+	switch statusCode {
+	case 409:
 		logBlockRejected(block, c.sanitizedURL, resp)
 
 		// Just drop block on 409 status code like Prometheus does.
@@ -461,22 +462,28 @@ again:
 		// - 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
-	} else if statusCode == 415 || statusCode == 400 {
-		if c.canDowngradeVMProto.Swap(false) {
-			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", c.sanitizedURL)
-			c.useVMProto.Store(false)
-		}
-
+	case 415, 400:
 		if encoding.IsZstd(block) {
 			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", c.sanitizedURL)
 
-			block = mustRepackBlockFromZstdToSnappy(block)
+			zstdBlockLen := len(block)
+			block, err = repackBlockFromZstdToSnappy(block)
+			if err == nil {
+				if c.canDowngradeVMProto.Swap(false) {
+					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", c.sanitizedURL)
+					c.useVMProto.Store(false)
+				}
 
-			c.retriesCount.Inc()
-			_ = resp.Body.Close()
-			goto again
+				c.retriesCount.Inc()
+				_ = resp.Body.Close()
+				goto again
+			}
+
+			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)
 		}
 
 		// Just drop snappy blocks on 400 or 415 status codes like Prometheus does.
@@ -538,14 +545,21 @@ func getRetryDuration(retryAfterDuration, retryDuration, maxRetryDuration time.D
 	return retryDuration
 }
 
-func mustRepackBlockFromZstdToSnappy(zstdBlock []byte) []byte {
+// repackBlockFromZstdToSnappy repacks the given zstd-compressed block to snappy-compressed block.
+//
+// The input block may be corrupted, for example, if vmagent was shut down ungracefully and
+// failed to properly update the persisted queue files. In such cases, zstd decompression
+// will fail and an error will be returned.
+//
+// For more details, see: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9417
+func repackBlockFromZstdToSnappy(zstdBlock []byte) ([]byte, error) {
 	plainBlock := make([]byte, 0, len(zstdBlock)*2)
 	plainBlock, err := zstd.Decompress(plainBlock, zstdBlock)
 	if err != nil {
-		logger.Panicf("FATAL: cannot re-pack block with size %d bytes from Zstd to Snappy: %s", len(zstdBlock), err)
+		return nil, fmt.Errorf("zstd: decompress: %s", err)
 	}
 
-	return snappy.Encode(nil, plainBlock)
+	return snappy.Encode(nil, plainBlock), nil
 }
 
 func logBlockRejected(block []byte, sanitizedURL string, resp *http.Response) {

app/vmagent/remotewrite/client_test.go

@@ -25,7 +25,7 @@ func TestCalculateRetryDuration(t *testing.T) {
 		expectMaxDuration := helper(expectMinDuration)
 		expectMinDuration = expectMinDuration - (1000 * time.Millisecond) // Avoid edge case when calculating time.Until(now)
 
-		if !(retryDuration >= expectMinDuration && retryDuration <= expectMaxDuration) {
+		if retryDuration < expectMinDuration || retryDuration > expectMaxDuration {
 			t.Fatalf(
 				"incorrect retry duration, want (ms): [%d, %d], got (ms): %d",
 				expectMinDuration.Milliseconds(), expectMaxDuration.Milliseconds(),
@@ -105,7 +105,10 @@ func TestRepackBlockFromZstdToSnappy(t *testing.T) {
 	expectedPlainBlock := []byte(`foobar`)
 
 	zstdBlock := encoding.CompressZSTDLevel(nil, expectedPlainBlock, 1)
-	snappyBlock := mustRepackBlockFromZstdToSnappy(zstdBlock)
+	snappyBlock, err := repackBlockFromZstdToSnappy(zstdBlock)
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
 
 	actualPlainBlock, err := snappy.Decode(nil, snappyBlock)
 	if err != nil {
@@ -116,3 +119,14 @@ func TestRepackBlockFromZstdToSnappy(t *testing.T) {
 		t.Fatalf("unexpected plain block; got %q; want %q", actualPlainBlock, expectedPlainBlock)
 	}
 }
+
+func TestRepackBlockFromZstdToSnappyInvalidBlock(t *testing.T) {
+	snappyBlock, err := repackBlockFromZstdToSnappy([]byte("invalid zstd block"))
+
+	if err == nil {
+		t.Fatalf("expected error for invalid zstd block; got nil")
+	}
+	if len(snappyBlock) != 0 {
+		t.Fatalf("expected empty snappy block; got %d bytes", len(snappyBlock))
+	}
+}

app/vmagent/remotewrite/pendingseries.go

@@ -14,7 +14,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
@@ -24,9 +24,10 @@ import (
 
 var (
 	flushInterval = flag.Duration("remoteWrite.flushInterval", time.Second, "Interval for flushing the data to remote storage. "+
-		"This option takes effect only when less than 10K data points per second are pushed to -remoteWrite.url")
+		"This option takes effect only when less than -remoteWrite.maxRowsPerBlock data points per -remoteWrite.flushInterval are pushed to -remoteWrite.url")
 	maxUnpackedBlockSize = flagutil.NewBytes("remoteWrite.maxBlockSize", 8*1024*1024, "The maximum block size to send to remote storage. Bigger blocks may improve performance at the cost of the increased memory usage. See also -remoteWrite.maxRowsPerBlock")
 	maxRowsPerBlock      = flag.Int("remoteWrite.maxRowsPerBlock", 10000, "The maximum number of samples to send in each block to remote storage. Higher number may improve performance at the cost of the increased memory usage. See also -remoteWrite.maxBlockSize")
+	maxMetadataPerBlock  = flag.Int("remoteWrite.maxMetadataPerBlock", 5000, "The maximum number of metadata to send in each block to remote storage. Higher number may improve performance at the cost of the increased memory usage. See also -remoteWrite.maxBlockSize")
 	vmProtoCompressLevel = flag.Int("remoteWrite.vmProtoCompressLevel", 0, "The compression level for VictoriaMetrics remote write protocol. "+
 		"Higher values reduce network traffic at the cost of higher CPU usage. Negative values reduce CPU usage at the cost of increased network traffic. "+
 		"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol")
@@ -60,9 +61,16 @@ func (ps *pendingSeries) MustStop() {
 	ps.periodicFlusherWG.Wait()
 }
 
-func (ps *pendingSeries) TryPush(tss []prompbmarshal.TimeSeries) bool {
+func (ps *pendingSeries) TryPushTimeSeries(tss []prompb.TimeSeries) bool {
 	ps.mu.Lock()
-	ok := ps.wr.tryPush(tss)
+	ok := ps.wr.tryPushTimeSeries(tss)
+	ps.mu.Unlock()
+	return ok
+}
+
+func (ps *pendingSeries) TryPushMetadata(mms []prompb.MetricMetadata) bool {
+	ps.mu.Lock()
+	ok := ps.wr.tryPushMetadata(mms)
 	ps.mu.Unlock()
 	return ok
 }
@@ -108,29 +116,37 @@ type writeRequest struct {
 	// How many decimal digits after point must be left before sending the writeRequest to fq.
 	roundDigits int
 
-	wr prompbmarshal.WriteRequest
+	wr prompb.WriteRequest
 
-	tss     []prompbmarshal.TimeSeries
-	labels  []prompbmarshal.Label
-	samples []prompbmarshal.Sample
+	tss     []prompb.TimeSeries
+	mms     []prompb.MetricMetadata
+	labels  []prompb.Label
+	samples []prompb.Sample
 
 	// buf holds labels data
 	buf []byte
+	// metadatabuf holds metadata data
+	metadatabuf []byte
 }
 
 func (wr *writeRequest) reset() {
 	// Do not reset lastFlushTime, fq, isVMRemoteWrite, significantFigures and roundDigits, since they are reused.
 
 	wr.wr.Timeseries = nil
+	wr.wr.Metadata = nil
 
 	clear(wr.tss)
 	wr.tss = wr.tss[:0]
 
+	clear(wr.mms)
+	wr.mms = wr.mms[:0]
+
 	promrelabel.CleanLabels(wr.labels)
 	wr.labels = wr.labels[:0]
 
 	wr.samples = wr.samples[:0]
 	wr.buf = wr.buf[:0]
+	wr.metadatabuf = wr.metadatabuf[:0]
 }
 
 // mustFlushOnStop force pushes wr data into wr.fq
@@ -138,6 +154,7 @@ func (wr *writeRequest) reset() {
 // This is needed in order to properly save in-memory data to persistent queue on graceful shutdown.
 func (wr *writeRequest) mustFlushOnStop() {
 	wr.wr.Timeseries = wr.tss
+	wr.wr.Metadata = wr.mms
 	if !tryPushWriteRequest(&wr.wr, wr.mustWriteBlock, wr.isVMRemoteWrite.Load()) {
 		logger.Panicf("BUG: final flush must always return true")
 	}
@@ -151,6 +168,7 @@ func (wr *writeRequest) mustWriteBlock(block []byte) bool {
 
 func (wr *writeRequest) tryFlush() bool {
 	wr.wr.Timeseries = wr.tss
+	wr.wr.Metadata = wr.mms
 	wr.lastFlushTime.Store(fasttime.UnixTimestamp())
 	if !tryPushWriteRequest(&wr.wr, wr.fq.TryWriteBlock, wr.isVMRemoteWrite.Load()) {
 		return false
@@ -159,7 +177,7 @@ func (wr *writeRequest) tryFlush() bool {
 	return true
 }
 
-func adjustSampleValues(samples []prompbmarshal.Sample, significantFigures, roundDigits int) {
+func adjustSampleValues(samples []prompb.Sample, significantFigures, roundDigits int) {
 	if n := significantFigures; n > 0 {
 		for i := range samples {
 			s := &samples[i]
@@ -174,7 +192,49 @@ func adjustSampleValues(samples []prompbmarshal.Sample, significantFigures, roun
 	}
 }
 
-func (wr *writeRequest) tryPush(src []prompbmarshal.TimeSeries) bool {
+func (wr *writeRequest) tryPushMetadata(mms []prompb.MetricMetadata) bool {
+	mmdDst := wr.mms
+	maxMetadataPerBlock := *maxMetadataPerBlock
+	for i := range mms {
+		if len(wr.mms) >= maxMetadataPerBlock {
+			if !wr.tryFlush() {
+				return false
+			}
+			mmdDst = wr.mms
+		}
+		mmSrc := &mms[i]
+		mmdDst = append(mmdDst, prompb.MetricMetadata{})
+		wr.copyMetadata(&mmdDst[len(mmdDst)-1], mmSrc)
+	}
+	wr.mms = mmdDst
+	return true
+}
+
+func (wr *writeRequest) copyMetadata(dst, src *prompb.MetricMetadata) {
+	// Direct copy for non-string fields, which are safe by value.
+	dst.Type = src.Type
+	dst.Unit = src.Unit
+
+	// Pre-allocate memory for all string fields.
+	neededBufLen := len(src.MetricFamilyName) + len(src.Help)
+	bufLen := len(wr.metadatabuf)
+	wr.metadatabuf = slicesutil.SetLength(wr.metadatabuf, bufLen+neededBufLen)
+	buf := wr.metadatabuf[:bufLen]
+
+	// Copy MetricFamilyName
+	bufLen = len(buf)
+	buf = append(buf, src.MetricFamilyName...)
+	dst.MetricFamilyName = bytesutil.ToUnsafeString(buf[bufLen:])
+
+	// Copy Help
+	bufLen = len(buf)
+	buf = append(buf, src.Help...)
+	dst.Help = bytesutil.ToUnsafeString(buf[bufLen:])
+
+	wr.metadatabuf = buf
+}
+
+func (wr *writeRequest) tryPushTimeSeries(src []prompb.TimeSeries) bool {
 	tssDst := wr.tss
 	maxSamplesPerBlock := *maxRowsPerBlock
 	// Allow up to 10x of labels per each block on average.
@@ -189,7 +249,7 @@ func (wr *writeRequest) tryPush(src []prompbmarshal.TimeSeries) bool {
 		}
 		tsSrc := &src[i]
 		adjustSampleValues(tsSrc.Samples, wr.significantFigures, wr.roundDigits)
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{})
+		tssDst = append(tssDst, prompb.TimeSeries{})
 		wr.copyTimeSeries(&tssDst[len(tssDst)-1], tsSrc)
 	}
 
@@ -197,7 +257,7 @@ func (wr *writeRequest) tryPush(src []prompbmarshal.TimeSeries) bool {
 	return true
 }
 
-func (wr *writeRequest) copyTimeSeries(dst, src *prompbmarshal.TimeSeries) {
+func (wr *writeRequest) copyTimeSeries(dst, src *prompb.TimeSeries) {
 	labelsSrc := src.Labels
 
 	// Pre-allocate memory for labels.
@@ -240,8 +300,8 @@ func (wr *writeRequest) copyTimeSeries(dst, src *prompbmarshal.TimeSeries) {
 // marshalConcurrency limits the maximum number of concurrent workers, which marshal and compress WriteRequest.
 var marshalConcurrencyCh = make(chan struct{}, cgroup.AvailableCPUs())
 
-func tryPushWriteRequest(wr *prompbmarshal.WriteRequest, tryPushBlock func(block []byte) bool, isVMRemoteWrite bool) bool {
-	if len(wr.Timeseries) == 0 {
+func tryPushWriteRequest(wr *prompb.WriteRequest, tryPushBlock func(block []byte) bool, isVMRemoteWrite bool) bool {
+	if wr.IsEmpty() {
 		// Nothing to push
 		return true
 	}
@@ -267,6 +327,7 @@ func tryPushWriteRequest(wr *prompbmarshal.WriteRequest, tryPushBlock func(block
 			compressBufPool.Put(zb)
 			if ok {
 				blockSizeRows.Update(float64(len(wr.Timeseries)))
+				blockMetadataRows.Update(float64(len(wr.Metadata)))
 				blockSizeBytes.Update(float64(zbLen))
 			}
 			return ok
@@ -278,47 +339,86 @@ func tryPushWriteRequest(wr *prompbmarshal.WriteRequest, tryPushBlock func(block
 		<-marshalConcurrencyCh
 	}
 
-	// Too big block. Recursively split it into smaller parts if possible.
-	if len(wr.Timeseries) == 1 {
-		// A single time series left. Recursively split its samples into smaller parts if possible.
+	// Split timeseries or metadata into two smaller blocks
+	switch len(wr.Timeseries) {
+	case 0:
+		if len(wr.Metadata) == 1 {
+			logger.Warnf("dropping a metadata exceeding -remoteWrite.maxBlockSize=%d bytes", maxUnpackedBlockSize.N)
+			return true
+		}
+		metadata := wr.Metadata
+		n := len(metadata) / 2
+		wr.Metadata = metadata[:n]
+		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
+			wr.Metadata = metadata
+			return false
+		}
+		wr.Metadata = metadata[n:]
+		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
+			wr.Metadata = metadata
+			return false
+		}
+		wr.Metadata = metadata
+		return true
+
+	case 1:
+		// A single time series left. Recursively split its samples and metadata into smaller parts if possible.
 		samples := wr.Timeseries[0].Samples
-		if len(samples) == 1 {
-			logger.Warnf("dropping a sample for metric with too long labels exceeding -remoteWrite.maxBlockSize=%d bytes", maxUnpackedBlockSize.N)
+		metaData := wr.Metadata
+		if len(samples) == 1 && len(metaData) <= 1 {
+			logger.Warnf("dropping a sample for metric and %d metadata which are exceeding -remoteWrite.maxBlockSize=%d bytes", len(metaData), maxUnpackedBlockSize.N)
 			return true
 		}
 		n := len(samples) / 2
+		m := len(metaData) / 2
 		wr.Timeseries[0].Samples = samples[:n]
+		wr.Metadata = metaData[:m]
 		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
 			wr.Timeseries[0].Samples = samples
+			wr.Metadata = metaData
 			return false
 		}
 		wr.Timeseries[0].Samples = samples[n:]
+		wr.Metadata = metaData[m:]
 		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
 			wr.Timeseries[0].Samples = samples
+			wr.Metadata = metaData
 			return false
 		}
 		wr.Timeseries[0].Samples = samples
+		wr.Metadata = metaData
+		return true
+
+	default:
+		// Split both timeseries and metadata.
+		timeseries := wr.Timeseries
+		metaData := wr.Metadata
+		n := len(timeseries) / 2
+		m := len(metaData) / 2
+		wr.Timeseries = timeseries[:n]
+		wr.Metadata = metaData[:m]
+		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
+			wr.Timeseries = timeseries
+			wr.Metadata = metaData
+			return false
+		}
+		wr.Timeseries = timeseries[n:]
+		wr.Metadata = metaData[m:]
+		if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
+			wr.Timeseries = timeseries
+			wr.Metadata = metaData
+			return false
+		}
+		wr.Timeseries = timeseries
+		wr.Metadata = metaData
 		return true
 	}
-	timeseries := wr.Timeseries
-	n := len(timeseries) / 2
-	wr.Timeseries = timeseries[:n]
-	if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
-		wr.Timeseries = timeseries
-		return false
-	}
-	wr.Timeseries = timeseries[n:]
-	if !tryPushWriteRequest(wr, tryPushBlock, isVMRemoteWrite) {
-		wr.Timeseries = timeseries
-		return false
-	}
-	wr.Timeseries = timeseries
-	return true
 }
 
 var (
-	blockSizeBytes = metrics.NewHistogram(`vmagent_remotewrite_block_size_bytes`)
-	blockSizeRows  = metrics.NewHistogram(`vmagent_remotewrite_block_size_rows`)
+	blockSizeBytes    = metrics.NewHistogram(`vmagent_remotewrite_block_size_bytes`)
+	blockSizeRows     = metrics.NewHistogram(`vmagent_remotewrite_block_size_rows`)
+	blockMetadataRows = metrics.NewHistogram(`vmagent_remotewrite_block_metadata_rows`)
 )
 
 var (

app/vmagent/remotewrite/pendingseries_test.go

@@ -5,7 +5,7 @@ import (
 	"math"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 func TestPushWriteRequest(t *testing.T) {
@@ -49,19 +49,19 @@ func testPushWriteRequest(t *testing.T, rowsCount, expectedBlockLenProm, expecte
 	f(true, expectedBlockLenVM, 15)
 }
 
-func newTestWriteRequest(seriesCount, labelsCount int) *prompbmarshal.WriteRequest {
-	var wr prompbmarshal.WriteRequest
+func newTestWriteRequest(seriesCount, labelsCount int) *prompb.WriteRequest {
+	var wr prompb.WriteRequest
 	for i := 0; i < seriesCount; i++ {
-		var labels []prompbmarshal.Label
+		var labels []prompb.Label
 		for j := 0; j < labelsCount; j++ {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  fmt.Sprintf("label_%d_%d", i, j),
 				Value: fmt.Sprintf("value_%d_%d", i, j),
 			})
 		}
-		wr.Timeseries = append(wr.Timeseries, prompbmarshal.TimeSeries{
+		wr.Timeseries = append(wr.Timeseries, prompb.TimeSeries{
 			Labels: labels,
-			Samples: []prompbmarshal.Sample{
+			Samples: []prompb.Sample{
 				{
 					Value:     float64(i),
 					Timestamp: 1000 * int64(i),

app/vmagent/remotewrite/relabel.go

@@ -10,7 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 
 	"github.com/VictoriaMetrics/metrics"
@@ -32,7 +32,7 @@ var (
 		"See https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels")
 )
 
-var labelsGlobal []prompbmarshal.Label
+var labelsGlobal []prompb.Label
 
 var (
 	relabelConfigReloads      *metrics.Counter
@@ -146,14 +146,14 @@ func initLabelsGlobal() {
 		if n < 0 {
 			logger.Fatalf("missing '=' in `-remoteWrite.label`. It must contain label in the form `name=value`; got %q", s)
 		}
-		labelsGlobal = append(labelsGlobal, prompbmarshal.Label{
+		labelsGlobal = append(labelsGlobal, prompb.Label{
 			Name:  s[:n],
 			Value: s[n+1:],
 		})
 	}
 }
 
-func (rctx *relabelCtx) applyRelabeling(tss []prompbmarshal.TimeSeries, pcs *promrelabel.ParsedConfigs) []prompbmarshal.TimeSeries {
+func (rctx *relabelCtx) applyRelabeling(tss []prompb.TimeSeries, pcs *promrelabel.ParsedConfigs) []prompb.TimeSeries {
 	if pcs.Len() == 0 && !*usePromCompatibleNaming {
 		// Nothing to change.
 		return tss
@@ -174,7 +174,7 @@ func (rctx *relabelCtx) applyRelabeling(tss []prompbmarshal.TimeSeries, pcs *pro
 		if *usePromCompatibleNaming {
 			fixPromCompatibleNaming(labels[labelsLen:])
 		}
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: ts.Samples,
 		})
@@ -183,7 +183,7 @@ func (rctx *relabelCtx) applyRelabeling(tss []prompbmarshal.TimeSeries, pcs *pro
 	return tssDst
 }
 
-func (rctx *relabelCtx) appendExtraLabels(tss []prompbmarshal.TimeSeries, extraLabels []prompbmarshal.Label) {
+func (rctx *relabelCtx) appendExtraLabels(tss []prompb.TimeSeries, extraLabels []prompb.Label) {
 	if len(extraLabels) == 0 {
 		return
 	}
@@ -207,7 +207,7 @@ func (rctx *relabelCtx) appendExtraLabels(tss []prompbmarshal.TimeSeries, extraL
 	rctx.labels = labels
 }
 
-func (rctx *relabelCtx) tenantToLabels(tss []prompbmarshal.TimeSeries, accountID, projectID uint32) {
+func (rctx *relabelCtx) tenantToLabels(tss []prompb.TimeSeries, accountID, projectID uint32) {
 	rctx.reset()
 	accountIDStr := strconv.FormatUint(uint64(accountID), 10)
 	projectIDStr := strconv.FormatUint(uint64(projectID), 10)
@@ -222,11 +222,11 @@ func (rctx *relabelCtx) tenantToLabels(tss []prompbmarshal.TimeSeries, accountID
 			}
 			labels = append(labels, label)
 		}
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "vm_account_id",
 			Value: accountIDStr,
 		})
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "vm_project_id",
 			Value: projectIDStr,
 		})
@@ -237,7 +237,7 @@ func (rctx *relabelCtx) tenantToLabels(tss []prompbmarshal.TimeSeries, accountID
 
 type relabelCtx struct {
 	// pool for labels, which are used during the relabeling.
-	labels []prompbmarshal.Label
+	labels []prompb.Label
 }
 
 func (rctx *relabelCtx) reset() {
@@ -260,7 +260,7 @@ func putRelabelCtx(rctx *relabelCtx) {
 	relabelCtxPool.Put(rctx)
 }
 
-func fixPromCompatibleNaming(labels []prompbmarshal.Label) {
+func fixPromCompatibleNaming(labels []prompb.Label) {
 	// Replace unsupported Prometheus chars in label names and metric names with underscores.
 	for i := range labels {
 		label := &labels[i]

app/vmagent/remotewrite/relabel_test.go

@@ -4,7 +4,7 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
@@ -39,7 +39,7 @@ func TestApplyRelabeling(t *testing.T) {
 }
 
 func TestAppendExtraLabels(t *testing.T) {
-	f := func(extraLabels []prompbmarshal.Label, sTss, sExpTss string) {
+	f := func(extraLabels []prompb.Label, sTss, sExpTss string) {
 		t.Helper()
 		rctx := &relabelCtx{}
 		tss, expTss := parseSeries(sTss), parseSeries(sExpTss)
@@ -50,19 +50,19 @@ func TestAppendExtraLabels(t *testing.T) {
 	}
 
 	f(nil, "up", "up")
-	f([]prompbmarshal.Label{{Name: "foo", Value: "bar"}}, "up", `up{foo="bar"}`)
-	f([]prompbmarshal.Label{{Name: "foo", Value: "bar"}}, `up{foo="baz"}`, `up{foo="bar"}`)
-	f([]prompbmarshal.Label{{Name: "baz", Value: "qux"}}, `up{foo="baz"}`, `up{foo="baz",baz="qux"}`)
+	f([]prompb.Label{{Name: "foo", Value: "bar"}}, "up", `up{foo="bar"}`)
+	f([]prompb.Label{{Name: "foo", Value: "bar"}}, `up{foo="baz"}`, `up{foo="bar"}`)
+	f([]prompb.Label{{Name: "baz", Value: "qux"}}, `up{foo="baz"}`, `up{foo="baz",baz="qux"}`)
 
 	oldVal := *usePromCompatibleNaming
 	*usePromCompatibleNaming = true
-	f([]prompbmarshal.Label{{Name: "foo.bar", Value: "baz"}}, "up", `up{foo.bar="baz"}`)
+	f([]prompb.Label{{Name: "foo.bar", Value: "baz"}}, "up", `up{foo.bar="baz"}`)
 	*usePromCompatibleNaming = oldVal
 }
 
-func parseSeries(data string) []prompbmarshal.TimeSeries {
-	var tss []prompbmarshal.TimeSeries
-	tss = append(tss, prompbmarshal.TimeSeries{
+func parseSeries(data string) []prompb.TimeSeries {
+	var tss []prompb.TimeSeries
+	tss = append(tss, prompb.TimeSeries{
 		Labels: promutil.MustNewLabelsFromString(data).GetLabels(),
 	})
 	return tss

app/vmagent/remotewrite/remotewrite.go

@@ -23,7 +23,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
@@ -209,7 +209,7 @@ func Init() {
 	// In this case it is impossible to prevent from sending many duplicates of samples passed to TryPush() to all the configured -remoteWrite.url
 	// if these samples couldn't be sent to the -remoteWrite.url with the disabled persistent queue. So it is better sending samples
 	// to the remaining -remoteWrite.url and dropping them on the blocked queue.
-	dropSamplesOnFailureGlobal = *dropSamplesOnOverload || disableOnDiskQueueAny && len(disableOnDiskQueues) > 1
+	dropSamplesOnFailureGlobal = *dropSamplesOnOverload || disableOnDiskQueueAny && len(*remoteWriteURLs) > 1
 
 	dropDanglingQueues()
 
@@ -253,7 +253,7 @@ func dropDanglingQueues() {
 		if _, ok := existingQueues[dirname]; !ok {
 			logger.Infof("removing dangling queue %q", dirname)
 			fullPath := filepath.Join(queuesDir, dirname)
-			fs.MustRemoveAll(fullPath)
+			fs.MustRemoveDir(fullPath)
 			removed++
 		}
 	}
@@ -372,7 +372,7 @@ func Stop() {
 // PushDropSamplesOnFailure drops wr samples if they cannot be sent to -remoteWrite.url by any reason.
 //
 // PushDropSamplesOnFailure can modify wr contents.
-func PushDropSamplesOnFailure(at *auth.Token, wr *prompbmarshal.WriteRequest) {
+func PushDropSamplesOnFailure(at *auth.Token, wr *prompb.WriteRequest) {
 	_ = tryPush(at, wr, true)
 }
 
@@ -382,19 +382,13 @@ func PushDropSamplesOnFailure(at *auth.Token, wr *prompbmarshal.WriteRequest) {
 // TryPush may send partial data from wr on unsuccessful attempt, so repeated call for the same wr may send the data multiple times.
 //
 // The caller must return ErrQueueFullHTTPRetry to the client, which sends wr, if TryPush returns false.
-func TryPush(at *auth.Token, wr *prompbmarshal.WriteRequest) bool {
+func TryPush(at *auth.Token, wr *prompb.WriteRequest) bool {
 	return tryPush(at, wr, dropSamplesOnFailureGlobal)
 }
 
-func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnFailure bool) bool {
+func tryPush(at *auth.Token, wr *prompb.WriteRequest, forceDropSamplesOnFailure bool) bool {
 	tss := wr.Timeseries
-
-	var tenantRctx *relabelCtx
-	if at != nil {
-		// Convert at to (vm_account_id, vm_project_id) labels.
-		tenantRctx = getRelabelCtx()
-		defer putRelabelCtx(tenantRctx)
-	}
+	mms := wr.Metadata
 
 	// Quick check whether writes to configured remote storage systems are blocked.
 	// This allows saving CPU time spent on relabeling and block compression
@@ -411,6 +405,23 @@ func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnF
 		return true
 	}
 
+	// Push metadata separately from time series, since it doesn't need sharding,
+	// relabeling, stream aggregation, deduplication, etc.
+	if !tryPushMetadataToRemoteStorages(rwctxs, mms, forceDropSamplesOnFailure) {
+		return false
+	}
+
+	if len(tss) == 0 {
+		return true
+	}
+
+	var tenantRctx *relabelCtx
+	if at != nil {
+		// Convert at to (vm_account_id, vm_project_id) labels.
+		tenantRctx = getRelabelCtx()
+		defer putRelabelCtx(tenantRctx)
+	}
+
 	var rctx *relabelCtx
 	rcs := allRelabelConfigs.Load()
 	pcsGlobal := rcs.global
@@ -481,7 +492,7 @@ func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnF
 			deduplicatorGlobal.Push(tssBlock)
 			tssBlock = tssBlock[:0]
 		}
-		if !tryPushBlockToRemoteStorages(rwctxs, tssBlock, forceDropSamplesOnFailure) {
+		if !tryPushTimeSeriesToRemoteStorages(rwctxs, tssBlock, forceDropSamplesOnFailure) {
 			return false
 		}
 	}
@@ -492,7 +503,7 @@ func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnF
 // returns only the unblocked rwctx.
 //
 // calculateHealthyRwctxIdx will rely on the order of rwctx to be in ascending order.
-func getEligibleRemoteWriteCtxs(tss []prompbmarshal.TimeSeries, forceDropSamplesOnFailure bool) ([]*remoteWriteCtx, bool) {
+func getEligibleRemoteWriteCtxs(tss []prompb.TimeSeries, forceDropSamplesOnFailure bool) ([]*remoteWriteCtx, bool) {
 	if !disableOnDiskQueueAny {
 		return rwctxsGlobal, true
 	}
@@ -520,18 +531,49 @@ func getEligibleRemoteWriteCtxs(tss []prompbmarshal.TimeSeries, forceDropSamples
 	return rwctxs, true
 }
 
-func pushToRemoteStoragesTrackDropped(tss []prompbmarshal.TimeSeries) {
+func pushTimeSeriesToRemoteStoragesTrackDropped(tss []prompb.TimeSeries) {
 	rwctxs, _ := getEligibleRemoteWriteCtxs(tss, true)
 	if len(rwctxs) == 0 {
 		return
 	}
 
-	if !tryPushBlockToRemoteStorages(rwctxs, tss, true) {
-		logger.Panicf("BUG: tryPushBlockToRemoteStorages() must return true when forceDropSamplesOnFailure=true")
+	if !tryPushTimeSeriesToRemoteStorages(rwctxs, tss, true) {
+		logger.Panicf("BUG: tryPushTimeSeriesToRemoteStorages() must return true when forceDropSamplesOnFailure=true")
 	}
 }
 
-func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmarshal.TimeSeries, forceDropSamplesOnFailure bool) bool {
+func tryPushMetadataToRemoteStorages(rwctxs []*remoteWriteCtx, mms []prompb.MetricMetadata, forceDropSamplesOnFailure bool) bool {
+	if len(mms) == 0 {
+		// Nothing to push
+		return true
+	}
+	// 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.
+	//
+	// Push metadata to remote storage systems in parallel to reduce
+	// the time needed for sending the data to multiple remote storage systems.
+	var wg sync.WaitGroup
+	wg.Add(len(rwctxs))
+	var anyPushFailed atomic.Bool
+	for _, rwctx := range rwctxs {
+		go func(rwctx *remoteWriteCtx) {
+			defer wg.Done()
+			if !rwctx.tryPushMetadataInternal(mms) {
+				rwctx.pushFailures.Inc()
+				if forceDropSamplesOnFailure {
+					rwctx.metadataDroppedOnPushFailure.Add(len(mms))
+					return
+				}
+				anyPushFailed.Store(true)
+			}
+		}(rwctx)
+	}
+	wg.Wait()
+	return !anyPushFailed.Load()
+}
+
+func tryPushTimeSeriesToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompb.TimeSeries, forceDropSamplesOnFailure bool) bool {
 	if len(tssBlock) == 0 {
 		// Nothing to push
 		return true
@@ -539,7 +581,7 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 
 	if len(rwctxs) == 1 {
 		// Fast path - just push data to the configured single remote storage
-		return rwctxs[0].TryPush(tssBlock, forceDropSamplesOnFailure)
+		return rwctxs[0].TryPushTimeSeries(tssBlock, forceDropSamplesOnFailure)
 	}
 
 	// We need to push tssBlock to multiple remote storages.
@@ -550,11 +592,11 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 		if replicas <= 0 {
 			replicas = 1
 		}
-		return tryShardingBlockAmongRemoteStorages(rwctxs, tssBlock, replicas, forceDropSamplesOnFailure)
+		return tryShardingTimeSeriesAmongRemoteStorages(rwctxs, tssBlock, replicas, forceDropSamplesOnFailure)
 	}
 
 	// Replicate tssBlock samples among rwctxs.
-	// Push tssBlock to remote storage systems in parallel in order to reduce
+	// Push tssBlock to remote storage systems in parallel to reduce
 	// the time needed for sending the data to multiple remote storage systems.
 	var wg sync.WaitGroup
 	wg.Add(len(rwctxs))
@@ -562,7 +604,7 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 	for _, rwctx := range rwctxs {
 		go func(rwctx *remoteWriteCtx) {
 			defer wg.Done()
-			if !rwctx.TryPush(tssBlock, forceDropSamplesOnFailure) {
+			if !rwctx.TryPushTimeSeries(tssBlock, forceDropSamplesOnFailure) {
 				anyPushFailed.Store(true)
 			}
 		}(rwctx)
@@ -571,7 +613,7 @@ func tryPushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmar
 	return !anyPushFailed.Load()
 }
 
-func tryShardingBlockAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmarshal.TimeSeries, replicas int, forceDropSamplesOnFailure bool) bool {
+func tryShardingTimeSeriesAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompb.TimeSeries, replicas int, forceDropSamplesOnFailure bool) bool {
 	x := getTSSShards(len(rwctxs))
 	defer putTSSShards(x)
 
@@ -588,9 +630,9 @@ func tryShardingBlockAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []pr
 			continue
 		}
 		wg.Add(1)
-		go func(rwctx *remoteWriteCtx, tss []prompbmarshal.TimeSeries) {
+		go func(rwctx *remoteWriteCtx, tss []prompb.TimeSeries) {
 			defer wg.Done()
-			if !rwctx.TryPush(tss, forceDropSamplesOnFailure) {
+			if !rwctx.TryPushTimeSeries(tss, forceDropSamplesOnFailure) {
 				anyPushFailed.Store(true)
 			}
 		}(rwctx, shard)
@@ -624,7 +666,7 @@ func calculateHealthyRwctxIdx(healthyRwctxs []*remoteWriteCtx) ([]int, []int) {
 }
 
 // shardAmountRemoteWriteCtx distribute time series to shards by consistent hashing.
-func shardAmountRemoteWriteCtx(tssBlock []prompbmarshal.TimeSeries, shards [][]prompbmarshal.TimeSeries, rwctxs []*remoteWriteCtx, replicas int) {
+func shardAmountRemoteWriteCtx(tssBlock []prompb.TimeSeries, shards [][]prompb.TimeSeries, rwctxs []*remoteWriteCtx, replicas int) {
 	tmpLabels := promutil.GetLabels()
 	defer promutil.PutLabels(tmpLabels)
 
@@ -680,7 +722,7 @@ func shardAmountRemoteWriteCtx(tssBlock []prompbmarshal.TimeSeries, shards [][]p
 }
 
 type tssShards struct {
-	shards [][]prompbmarshal.TimeSeries
+	shards [][]prompb.TimeSeries
 }
 
 func getTSSShards(n int) *tssShards {
@@ -690,7 +732,7 @@ func getTSSShards(n int) *tssShards {
 	}
 	x := v.(*tssShards)
 	if cap(x.shards) < n {
-		x.shards = make([][]prompbmarshal.TimeSeries, n)
+		x.shards = make([][]prompb.TimeSeries, n)
 	}
 	x.shards = x.shards[:n]
 	return x
@@ -708,7 +750,7 @@ func putTSSShards(x *tssShards) {
 var tssShardsPool sync.Pool
 
 // sortLabelsIfNeeded sorts labels if -sortLabels command-line flag is set.
-func sortLabelsIfNeeded(tss []prompbmarshal.TimeSeries) {
+func sortLabelsIfNeeded(tss []prompb.TimeSeries) {
 	if !*sortLabels {
 		return
 	}
@@ -717,11 +759,11 @@ func sortLabelsIfNeeded(tss []prompbmarshal.TimeSeries) {
 	}
 }
 
-func limitSeriesCardinality(tss []prompbmarshal.TimeSeries) []prompbmarshal.TimeSeries {
+func limitSeriesCardinality(tss []prompb.TimeSeries) []prompb.TimeSeries {
 	if hourlySeriesLimiter == nil && dailySeriesLimiter == nil {
 		return tss
 	}
-	dst := make([]prompbmarshal.TimeSeries, 0, len(tss))
+	dst := make([]prompb.TimeSeries, 0, len(tss))
 	for i := range tss {
 		labels := tss[i].Labels
 		h := getLabelsHash(labels)
@@ -748,7 +790,7 @@ var (
 	dailySeriesLimitRowsDropped  = metrics.NewCounter(`vmagent_daily_series_limit_rows_dropped_total`)
 )
 
-func getLabelsHash(labels []prompbmarshal.Label) uint64 {
+func getLabelsHash(labels []prompb.Label) uint64 {
 	bb := labelsHashBufPool.Get()
 	b := bb.B[:0]
 	for _, label := range labels {
@@ -763,12 +805,12 @@ func getLabelsHash(labels []prompbmarshal.Label) uint64 {
 
 var labelsHashBufPool bytesutil.ByteBufferPool
 
-func logSkippedSeries(labels []prompbmarshal.Label, flagName string, flagValue int) {
+func logSkippedSeries(labels []prompb.Label, flagName string, flagValue int) {
 	select {
 	case <-logSkippedSeriesTicker.C:
 		// Do not use logger.WithThrottler() here, since this will increase CPU usage
-		// because every call to logSkippedSeries will result to a call to prompbmarshal.LabelsToString.
-		logger.Warnf("skip series %s because %s=%d reached", prompbmarshal.LabelsToString(labels), flagName, flagValue)
+		// because every call to logSkippedSeries will result to a call to prompb.LabelsToString.
+		logger.Warnf("skip series %s because %s=%d reached", prompb.LabelsToString(labels), flagName, flagValue)
 	default:
 	}
 }
@@ -797,8 +839,9 @@ type remoteWriteCtx struct {
 	rowsPushedAfterRelabel *metrics.Counter
 	rowsDroppedByRelabel   *metrics.Counter
 
-	pushFailures             *metrics.Counter
-	rowsDroppedOnPushFailure *metrics.Counter
+	pushFailures                 *metrics.Counter
+	metadataDroppedOnPushFailure *metrics.Counter
+	rowsDroppedOnPushFailure     *metrics.Counter
 }
 
 func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks int, sanitizedURL string) *remoteWriteCtx {
@@ -862,8 +905,9 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
 		rowsPushedAfterRelabel: metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_rows_pushed_after_relabel_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
 		rowsDroppedByRelabel:   metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_relabel_metrics_dropped_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
 
-		pushFailures:             metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_push_failures_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
-		rowsDroppedOnPushFailure: metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_samples_dropped_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
+		pushFailures:                 metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_push_failures_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
+		metadataDroppedOnPushFailure: metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_metadata_dropped_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
+		rowsDroppedOnPushFailure:     metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_samples_dropped_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
 	}
 	rwctx.initStreamAggrConfig()
 
@@ -897,17 +941,17 @@ func (rwctx *remoteWriteCtx) MustStop() {
 	rwctx.rowsDroppedByRelabel = nil
 }
 
-// TryPush sends tss series to the configured remote write endpoint
+// TryPushTimeSeries sends tss series to the configured remote write endpoint
 //
-// TryPush doesn't modify tss, so tss can be passed concurrently to TryPush across distinct rwctx instances.
-func (rwctx *remoteWriteCtx) TryPush(tss []prompbmarshal.TimeSeries, forceDropSamplesOnFailure bool) bool {
+// TryPushTimeSeries doesn't modify tss, so tss can be passed concurrently to TryPush across distinct rwctx instances.
+func (rwctx *remoteWriteCtx) TryPushTimeSeries(tss []prompb.TimeSeries, forceDropSamplesOnFailure bool) bool {
 	var rctx *relabelCtx
-	var v *[]prompbmarshal.TimeSeries
+	var v *[]prompb.TimeSeries
 	defer func() {
 		if rctx == nil {
 			return
 		}
-		*v = prompbmarshal.ResetTimeSeries(tss)
+		*v = prompb.ResetTimeSeries(tss)
 		tssPool.Put(v)
 		putRelabelCtx(rctx)
 	}()
@@ -921,7 +965,7 @@ func (rwctx *remoteWriteCtx) TryPush(tss []prompbmarshal.TimeSeries, forceDropSa
 		// from affecting time series for other remoteWrite.url configs.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/467
 		// and https://github.com/VictoriaMetrics/VictoriaMetrics/issues/599
-		v = tssPool.Get().(*[]prompbmarshal.TimeSeries)
+		v = tssPool.Get().(*[]prompb.TimeSeries)
 		tss = append(*v, tss...)
 		rowsCountBeforeRelabel := getRowsCount(tss)
 		tss = rctx.applyRelabeling(tss, pcs)
@@ -940,7 +984,7 @@ func (rwctx *remoteWriteCtx) TryPush(tss []prompbmarshal.TimeSeries, forceDropSa
 			if rctx == nil {
 				rctx = getRelabelCtx()
 				// Make a copy of tss before dropping aggregated series
-				v = tssPool.Get().(*[]prompbmarshal.TimeSeries)
+				v = tssPool.Get().(*[]prompb.TimeSeries)
 				tss = append(*v, tss...)
 			}
 			tss = dropAggregatedSeries(tss, matchIdxs.B, rwctx.streamAggrDropInput)
@@ -953,7 +997,7 @@ func (rwctx *remoteWriteCtx) TryPush(tss []prompbmarshal.TimeSeries, forceDropSa
 	}
 
 	// Try pushing tss to remote storage
-	if rwctx.tryPushInternal(tss) {
+	if rwctx.tryPushTimeSeriesInternal(tss) {
 		return true
 	}
 
@@ -969,7 +1013,7 @@ func (rwctx *remoteWriteCtx) TryPush(tss []prompbmarshal.TimeSeries, forceDropSa
 
 var matchIdxsPool bytesutil.ByteBufferPool
 
-func dropAggregatedSeries(src []prompbmarshal.TimeSeries, matchIdxs []byte, dropInput bool) []prompbmarshal.TimeSeries {
+func dropAggregatedSeries(src []prompb.TimeSeries, matchIdxs []byte, dropInput bool) []prompb.TimeSeries {
 	dst := src[:0]
 	if !dropInput {
 		for i, match := range matchIdxs {
@@ -984,8 +1028,8 @@ func dropAggregatedSeries(src []prompbmarshal.TimeSeries, matchIdxs []byte, drop
 	return dst
 }
 
-func (rwctx *remoteWriteCtx) pushInternalTrackDropped(tss []prompbmarshal.TimeSeries) {
-	if rwctx.tryPushInternal(tss) {
+func (rwctx *remoteWriteCtx) pushInternalTrackDropped(tss []prompb.TimeSeries) {
+	if rwctx.tryPushTimeSeriesInternal(tss) {
 		return
 	}
 	if !rwctx.fq.IsPersistentQueueDisabled() {
@@ -996,14 +1040,21 @@ func (rwctx *remoteWriteCtx) pushInternalTrackDropped(tss []prompbmarshal.TimeSe
 	rwctx.rowsDroppedOnPushFailure.Add(rowsCount)
 }
 
-func (rwctx *remoteWriteCtx) tryPushInternal(tss []prompbmarshal.TimeSeries) bool {
+func (rwctx *remoteWriteCtx) tryPushMetadataInternal(mms []prompb.MetricMetadata) bool {
+	pss := rwctx.pss
+	idx := rwctx.pssNextIdx.Add(1) % uint64(len(pss))
+
+	return pss[idx].TryPushMetadata(mms)
+}
+
+func (rwctx *remoteWriteCtx) tryPushTimeSeriesInternal(tss []prompb.TimeSeries) bool {
 	var rctx *relabelCtx
-	var v *[]prompbmarshal.TimeSeries
+	var v *[]prompb.TimeSeries
 	defer func() {
 		if rctx == nil {
 			return
 		}
-		*v = prompbmarshal.ResetTimeSeries(tss)
+		*v = prompb.ResetTimeSeries(tss)
 		tssPool.Put(v)
 		putRelabelCtx(rctx)
 	}()
@@ -1012,7 +1063,7 @@ func (rwctx *remoteWriteCtx) tryPushInternal(tss []prompbmarshal.TimeSeries) boo
 		// Make a copy of tss before adding extra labels in order to prevent
 		// from affecting time series for other remoteWrite.url configs.
 		rctx = getRelabelCtx()
-		v = tssPool.Get().(*[]prompbmarshal.TimeSeries)
+		v = tssPool.Get().(*[]prompb.TimeSeries)
 		tss = append(*v, tss...)
 		rctx.appendExtraLabels(tss, labelsGlobal)
 	}
@@ -1020,17 +1071,17 @@ func (rwctx *remoteWriteCtx) tryPushInternal(tss []prompbmarshal.TimeSeries) boo
 	pss := rwctx.pss
 	idx := rwctx.pssNextIdx.Add(1) % uint64(len(pss))
 
-	return pss[idx].TryPush(tss)
+	return pss[idx].TryPushTimeSeries(tss)
 }
 
 var tssPool = &sync.Pool{
 	New: func() any {
-		a := []prompbmarshal.TimeSeries{}
+		a := []prompb.TimeSeries{}
 		return &a
 	},
 }
 
-func getRowsCount(tss []prompbmarshal.TimeSeries) int {
+func getRowsCount(tss []prompb.TimeSeries) int {
 	rowsCount := 0
 	for _, ts := range tss {
 		rowsCount += len(ts.Samples)

app/vmagent/remotewrite/remotewrite_test.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/consistenthash"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
@@ -25,14 +25,14 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 		// Distribute itemsCount hashes returned by getLabelsHash() across bucketsCount buckets.
 		itemsCount := 1_000 * bucketsCount
 		m := make([]int, bucketsCount)
-		var labels []prompbmarshal.Label
+		var labels []prompb.Label
 		for i := 0; i < itemsCount; i++ {
-			labels = append(labels[:0], prompbmarshal.Label{
+			labels = append(labels[:0], prompb.Label{
 				Name:  "__name__",
 				Value: fmt.Sprintf("some_name_%d", i),
 			})
 			for j := 0; j < 10; j++ {
-				labels = append(labels, prompbmarshal.Label{
+				labels = append(labels, prompb.Label{
 					Name:  fmt.Sprintf("label_%d", j),
 					Value: fmt.Sprintf("value_%d_%d", i, j),
 				})
@@ -88,7 +88,7 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 		}
 
 		if streamAggrConfig != "" {
-			pushNoop := func(_ []prompbmarshal.TimeSeries) {}
+			pushNoop := func(_ []prompb.TimeSeries) {}
 			opts := streamaggr.Options{
 				EnableWindows: enableWindows,
 			}
@@ -102,11 +102,11 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 
 		offsetMsecs := time.Now().UnixMilli()
 		inputTss := prometheus.MustParsePromMetrics(input, offsetMsecs)
-		expectedTss := make([]prompbmarshal.TimeSeries, len(inputTss))
+		expectedTss := make([]prompb.TimeSeries, len(inputTss))
 
 		// copy inputTss to make sure it is not mutated during TryPush call
 		copy(expectedTss, inputTss)
-		if !rwctx.TryPush(inputTss, false) {
+		if !rwctx.TryPushTimeSeries(inputTss, false) {
 			t.Fatalf("cannot push samples to rwctx")
 		}
 
@@ -220,16 +220,16 @@ func TestShardAmountRemoteWriteCtx(t *testing.T) {
 
 		seriesCount := 100000
 		// build 1000000 series
-		tssBlock := make([]prompbmarshal.TimeSeries, 0, seriesCount)
+		tssBlock := make([]prompb.TimeSeries, 0, seriesCount)
 		for i := 0; i < seriesCount; i++ {
-			tssBlock = append(tssBlock, prompbmarshal.TimeSeries{
-				Labels: []prompbmarshal.Label{
+			tssBlock = append(tssBlock, prompb.TimeSeries{
+				Labels: []prompb.Label{
 					{
 						Name:  "label",
 						Value: strconv.Itoa(i),
 					},
 				},
-				Samples: []prompbmarshal.Sample{
+				Samples: []prompb.Sample{
 					{
 						Timestamp: 0,
 						Value:     0,
@@ -258,7 +258,7 @@ func TestShardAmountRemoteWriteCtx(t *testing.T) {
 		for i, nodeIdx := range healthyIdx {
 			for _, ts := range shards[i] {
 				// add it to node[nodeIdx]'s active time series
-				activeTimeSeriesByNodes[nodeIdx][prompbmarshal.LabelsToString(ts.Labels)] = struct{}{}
+				activeTimeSeriesByNodes[nodeIdx][prompb.LabelsToString(ts.Labels)] = struct{}{}
 			}
 		}
 
@@ -281,7 +281,7 @@ func TestShardAmountRemoteWriteCtx(t *testing.T) {
 		for i, nodeIdx := range healthyIdx {
 			for _, ts := range shards[i] {
 				// add it to node[nodeIdx]'s active time series
-				activeTimeSeriesByNodes[nodeIdx][prompbmarshal.LabelsToString(ts.Labels)] = struct{}{}
+				activeTimeSeriesByNodes[nodeIdx][prompb.LabelsToString(ts.Labels)] = struct{}{}
 			}
 		}
 

app/vmagent/remotewrite/streamaggr.go

@@ -8,7 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -80,7 +80,7 @@ func CheckStreamAggrConfigs() error {
 		return fmt.Errorf("too many -remoteWrite.streamAggr.config args: %d; it mustn't exceed the number of -remoteWrite.url args: %d", len(*streamAggrConfig), len(*remoteWriteURLs))
 	}
 
-	pushNoop := func(_ []prompbmarshal.TimeSeries) {}
+	pushNoop := func(_ []prompb.TimeSeries) {}
 	for idx := range *streamAggrConfig {
 		sas, err := newStreamAggrConfigPerURL(idx, pushNoop)
 		if err != nil {
@@ -141,7 +141,7 @@ func initStreamAggrConfigGlobal() {
 	}
 	dedupInterval := *streamAggrGlobalDedupInterval
 	if dedupInterval > 0 {
-		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
+		deduplicatorGlobal = streamaggr.NewDeduplicator(pushTimeSeriesToRemoteStoragesTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
 	}
 }
 
@@ -216,7 +216,7 @@ func newStreamAggrConfigGlobal() (*streamaggr.Aggregators, error) {
 		EnableWindows:        *streamAggrGlobalEnableWindows,
 	}
 
-	sas, err := streamaggr.LoadFromFile(path, pushToRemoteStoragesTrackDropped, opts, "global")
+	sas, err := streamaggr.LoadFromFile(path, pushTimeSeriesToRemoteStoragesTrackDropped, opts, "global")
 	if err != nil {
 		return nil, fmt.Errorf("cannot load -streamAggr.config=%q: %w", *streamAggrGlobalConfig, err)
 	}

app/vmagent/vmimport/request_handler.go

@@ -8,7 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport/stream"
@@ -36,7 +36,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	})
 }
 
-func insertRows(at *auth.Token, rows []vmimport.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []vmimport.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 
@@ -50,7 +50,7 @@ func insertRows(at *auth.Token, rows []vmimport.Row, extraLabels []prompbmarshal
 		labelsLen := len(labels)
 		for j := range r.Tags {
 			tag := &r.Tags[j]
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  bytesutil.ToUnsafeString(tag.Key),
 				Value: bytesutil.ToUnsafeString(tag.Value),
 			})
@@ -63,12 +63,12 @@ func insertRows(at *auth.Token, rows []vmimport.Row, extraLabels []prompbmarshal
 		}
 		samplesLen := len(samples)
 		for j, value := range values {
-			samples = append(samples, prompbmarshal.Sample{
+			samples = append(samples, prompb.Sample{
 				Value:     value,
 				Timestamp: timestamps[j],
 			})
 		}
-		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+		tssDst = append(tssDst, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})

app/vmalert-tool/unittest/recording.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/metricsql"
 )
@@ -49,7 +49,7 @@ Outer:
 		}
 		var expSamples []parsedSample
 		for _, s := range mt.ExpSamples {
-			expLb := []prompbmarshal.Label{}
+			expLb := []prompb.Label{}
 			if s.Labels != "" {
 				metricsqlExpr, err := metricsql.Parse(s.Labels)
 				if err != nil {
@@ -65,7 +65,7 @@ Outer:
 				}
 				if len(metricsqlMetricExpr.LabelFilterss) > 0 {
 					for _, l := range metricsqlMetricExpr.LabelFilterss[0] {
-						expLb = append(expLb, prompbmarshal.Label{
+						expLb = append(expLb, prompb.Label{
 							Name:  l.Label,
 							Value: l.Value,
 						})

app/vmalert-tool/unittest/testdata/rules.yaml

@@ -15,6 +15,10 @@ groups:
       - alert: SameAlertNameWithDifferentGroup
         expr: absent(test)
         for: 1m
+      - alert: AlertWithTemplate
+        expr: test
+        annotations:
+          queryAnno: '{{ query "foo" | first | value }}'
 
   - name: group2
     rules:

app/vmalert-tool/unittest/testdata/test1.yaml

@@ -10,7 +10,9 @@ tests:
     input_series:
       - series: "test"
         values: "_x5 1x5 _ stale"
-
+      - series: "foo"
+        values: "1x20"
+  
     alert_rule_test:
       - eval_time: 1m
         groupname: group1
@@ -32,6 +34,14 @@ tests:
         groupname: group1
         alertname: SameAlertNameWithDifferentGroup
         exp_alerts: []
+      - eval_time: 6m
+        groupname: group1
+        alertname: AlertWithTemplate
+        exp_alerts:
+          - exp_labels:
+              cluster: prod
+            exp_annotations:
+              queryAnno: '1'
 
     metricsql_expr_test:
       - expr: test
@@ -50,6 +60,8 @@ tests:
         values: "0+0x1440"
       - series: "test"
         values: "0+1x1440"
+      - series: "foo"
+        values: "1x20"
 
     metricsql_expr_test:
       - expr: count(ALERTS) by (alertgroup, alertname, alertstate)
@@ -59,6 +71,8 @@ tests:
             value: 1
           - labels: '{alertgroup="group1", alertname="InstanceDown", alertstate="pending"}'
             value: 1
+          - labels: '{alertgroup="group1", alertname="AlertWithTemplate", alertstate="firing"}'
+            value: 1
       - expr: t1
         eval_time: 4m
         exp_samples:

app/vmalert-tool/unittest/unittest.go

@@ -19,7 +19,6 @@ import (
 
 	"gopkg.in/yaml.v2"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	vmalertconfig "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -107,12 +106,12 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	vminsert.Init()
 	vmselect.Init()
 	// storagePath will be created again when closing vmselect, so remove it again.
-	defer fs.MustRemoveAll(storagePath)
+	defer fs.MustRemoveDir(storagePath)
 	defer vminsert.Stop()
 	defer vmselect.Stop()
 	disableAlertgroupLabel = disableGroupLabel
 
-	testfiles, err := config.ReadFromFS(files)
+	testfiles, err := vmalertconfig.ReadFromFS(files)
 	if err != nil {
 		logger.Fatalf("failed to load test files %q: %v", files, err)
 	}
@@ -304,7 +303,7 @@ checkCheck:
 func tearDown() {
 	vmstorage.Stop()
 	metrics.UnregisterAllMetrics()
-	fs.MustRemoveAll(storagePath)
+	fs.MustRemoveDir(storagePath)
 }
 
 func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]int, testGroups []vmalertconfig.Group, externalLabels map[string]string) (checkErrs []error) {
@@ -367,6 +366,7 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 			mergedExternalLabels[k] = v
 		}
 		ng := rule.NewGroup(group, q, time.Minute, mergedExternalLabels)
+		ng.Init()
 		groups = append(groups, ng)
 	}
 

app/vmalert/config/config.go

@@ -295,10 +295,7 @@ func parse(files map[string][]byte, validateTplFn ValidateTplFn, validateExpress
 }
 
 func parseConfig(data []byte) ([]Group, error) {
-	data, err := envtemplate.ReplaceBytes(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot expand environment vars: %w", err)
-	}
+	data = envtemplate.ReplaceBytes(data)
 
 	var result []Group
 	type cfgFile struct {
@@ -310,13 +307,13 @@ func parseConfig(data []byte) ([]Group, error) {
 	decoder := yaml.NewDecoder(bytes.NewReader(data))
 	for {
 		var cf cfgFile
-		if err = decoder.Decode(&cf); err != nil {
+		if err := decoder.Decode(&cf); err != nil {
 			if err == io.EOF { // EOF indicates no more documents to read
 				break
 			}
 			return nil, err
 		}
-		if err = checkOverflow(cf.XXX, "config"); err != nil {
+		if err := checkOverflow(cf.XXX, "config"); err != nil {
 			return nil, err
 		}
 		result = append(result, cf.Groups...)

app/vmalert/datasource/client_prom.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 
 	"github.com/valyala/fastjson"
 )
@@ -85,14 +85,14 @@ func (pi *promInstant) Unmarshal(b []byte) error {
 		labels := metric.GetObject()
 
 		r := &pi.ms[i]
-		r.Labels = make([]prompbmarshal.Label, 0, labels.Len())
+		r.Labels = make([]prompb.Label, 0, labels.Len())
 		labels.Visit(func(key []byte, v *fastjson.Value) {
 			lv, errLocal := v.StringBytes()
 			if errLocal != nil {
 				err = fmt.Errorf("error when parsing label value %q: %s", v, errLocal)
 				return
 			}
-			r.Labels = append(r.Labels, prompbmarshal.Label{
+			r.Labels = append(r.Labels, prompb.Label{
 				Name:  string(key),
 				Value: string(lv),
 			})

app/vmalert/datasource/client_test.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 var (
@@ -147,12 +147,12 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	expected := []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "vm_rows", Name: "__name__"}, {Value: "bar", Name: "foo"}},
+			Labels:     []prompb.Label{{Value: "vm_rows", Name: "__name__"}, {Value: "bar", Name: "foo"}},
 			Timestamps: []int64{1583786142},
 			Values:     []float64{13763},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Value: "vm_requests", Name: "__name__"}, {Value: "baz", Name: "foo"}},
+			Labels:     []prompb.Label{{Value: "vm_requests", Name: "__name__"}, {Value: "baz", Name: "foo"}},
 			Timestamps: []int64{1583786140},
 			Values:     []float64{2000},
 		},
@@ -225,7 +225,7 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	exp := []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "constantLine(10)", Name: "name"}},
+			Labels:     []prompb.Label{{Value: "constantLine(10)", Name: "name"}},
 			Timestamps: []int64{1611758403},
 			Values:     []float64{10},
 		},
@@ -247,12 +247,12 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	expected = []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}, {Value: "bar", Name: "foo"}},
+			Labels:     []prompb.Label{{Value: "total", Name: "stats_result"}, {Value: "bar", Name: "foo"}},
 			Timestamps: []int64{1583786142},
 			Values:     []float64{13763},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}, {Value: "baz", Name: "foo"}},
+			Labels:     []prompb.Label{{Value: "total", Name: "stats_result"}, {Value: "baz", Name: "foo"}},
 			Timestamps: []int64{1583786140},
 			Values:     []float64{2000},
 		},
@@ -455,7 +455,7 @@ func TestVMRangeQuery(t *testing.T) {
 		t.Fatalf("expected 1 metric  got %d in %+v", len(m), m)
 	}
 	expected := Metric{
-		Labels:     []prompbmarshal.Label{{Value: "vm_rows", Name: "__name__"}},
+		Labels:     []prompb.Label{{Value: "vm_rows", Name: "__name__"}},
 		Timestamps: []int64{1583786142},
 		Values:     []float64{13763},
 	}
@@ -486,7 +486,7 @@ func TestVMRangeQuery(t *testing.T) {
 		t.Fatalf("expected 1 metric  got %d in %+v", len(m), m)
 	}
 	expected = Metric{
-		Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}},
+		Labels:     []prompb.Label{{Value: "total", Name: "stats_result"}},
 		Timestamps: []int64{1583786142},
 		Values:     []float64{10},
 	}

app/vmalert/datasource/datasource.go

@@ -9,7 +9,7 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 // Querier interface wraps Query and QueryRange methods
@@ -60,7 +60,7 @@ type QuerierParams struct {
 
 // Metric is the basic entity which should be return by datasource
 type Metric struct {
-	Labels     []prompbmarshal.Label
+	Labels     []prompb.Label
 	Timestamps []int64
 	Values     []float64
 }
@@ -79,7 +79,7 @@ func (m *Metric) SetLabel(key, value string) {
 
 // AddLabel appends the given label to the label set
 func (m *Metric) AddLabel(key, value string) {
-	m.Labels = append(m.Labels, prompbmarshal.Label{Name: key, Value: value})
+	m.Labels = append(m.Labels, prompb.Label{Name: key, Value: value})
 }
 
 // DelLabel deletes the given label from the label set
@@ -103,7 +103,7 @@ func (m *Metric) Label(key string) string {
 }
 
 // Labels is collection of Label
-type Labels []prompbmarshal.Label
+type Labels []prompb.Label
 
 func (ls Labels) Len() int           { return len(ls) }
 func (ls Labels) Swap(i, j int)      { ls[i], ls[j] = ls[j], ls[i] }
@@ -158,7 +158,7 @@ func LabelCompare(a, b Labels) int {
 // ConvertToLabels convert map to Labels
 func ConvertToLabels(m map[string]string) (labelset Labels) {
 	for k, v := range m {
-		labelset = append(labelset, prompbmarshal.Label{
+		labelset = append(labelset, prompb.Label{
 			Name:  k,
 			Value: v,
 		})

app/vmalert/datasource/vm_prom_api_test.go

@@ -4,7 +4,7 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 func TestPromInstant_UnmarshalPositive(t *testing.T) {
@@ -23,7 +23,7 @@ func TestPromInstant_UnmarshalPositive(t *testing.T) {
 
 	f(`[{"metric":{"__name__":"up"},"value":[1583780000,"42"]}]`, []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "up"}},
+			Labels:     []prompb.Label{{Name: "__name__", Value: "up"}},
 			Timestamps: []int64{1583780000},
 			Values:     []float64{42},
 		},
@@ -33,17 +33,17 @@ func TestPromInstant_UnmarshalPositive(t *testing.T) {
 {"metric":{"__name__":"foo"},"value":[1583780001,"7"]},
 {"metric":{"__name__":"baz", "instance":"bar"},"value":[1583780002,"8"]}]`, []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "up"}},
+			Labels:     []prompb.Label{{Name: "__name__", Value: "up"}},
 			Timestamps: []int64{1583780000},
 			Values:     []float64{42},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "foo"}},
+			Labels:     []prompb.Label{{Name: "__name__", Value: "foo"}},
 			Timestamps: []int64{1583780001},
 			Values:     []float64{7},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "baz"}, {Name: "instance", Value: "bar"}},
+			Labels:     []prompb.Label{{Name: "__name__", Value: "baz"}, {Name: "instance", Value: "bar"}},
 			Timestamps: []int64{1583780002},
 			Values:     []float64{8},
 		},

app/vmalert/main_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/rule"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 )
 
@@ -108,8 +109,8 @@ groups:
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(f.Name()) }()
-	writeToFile(t, f.Name(), rules1)
+	defer fs.MustRemovePath(f.Name())
+	writeToFile(f.Name(), rules1)
 
 	*configCheckInterval = 200 * time.Millisecond
 	*rulePath = []string{f.Name()}
@@ -164,7 +165,7 @@ groups:
 		t.Fatalf("expected to have exactly 1 group loaded; got %d", groupsLen)
 	}
 
-	writeToFile(t, f.Name(), rules2)
+	writeToFile(f.Name(), rules2)
 	time.Sleep(*configCheckInterval * 2)
 	checkCfg(nil)
 	groupsLen = lenLocked(m)
@@ -172,7 +173,7 @@ groups:
 		t.Fatalf("expected to have exactly 2 groups loaded; got %d", groupsLen)
 	}
 
-	writeToFile(t, f.Name(), rules1)
+	writeToFile(f.Name(), rules1)
 	procutil.SelfSIGHUP()
 	time.Sleep(*configCheckInterval / 2)
 	checkCfg(nil)
@@ -181,7 +182,7 @@ groups:
 		t.Fatalf("expected to have exactly 1 group loaded; got %d", groupsLen)
 	}
 
-	writeToFile(t, f.Name(), `corrupted`)
+	writeToFile(f.Name(), `corrupted`)
 	procutil.SelfSIGHUP()
 	time.Sleep(*configCheckInterval / 2)
 	checkCfg(fmt.Errorf("config error"))
@@ -194,10 +195,6 @@ groups:
 	<-syncCh
 }
 
-func writeToFile(t *testing.T, file, b string) {
-	t.Helper()
-	err := os.WriteFile(file, []byte(b), 0644)
-	if err != nil {
-		t.Fatal(err)
-	}
+func writeToFile(file, b string) {
+	fs.MustWriteSync(file, []byte(b))
 }

app/vmalert/notifier/alert.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -195,10 +195,10 @@ func templateAnnotation(dst io.Writer, text string, data tplData, tpl *textTpl.T
 	return nil
 }
 
-func (a Alert) applyRelabelingIfNeeded(relabelCfg *promrelabel.ParsedConfigs) []prompbmarshal.Label {
-	var labels []prompbmarshal.Label
+func (a Alert) applyRelabelingIfNeeded(relabelCfg *promrelabel.ParsedConfigs) []prompb.Label {
+	var labels []prompb.Label
 	for k, v := range a.Labels {
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  promrelabel.SanitizeMetricName(k),
 			Value: v,
 		})

app/vmalert/notifier/alert_test.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -33,7 +33,7 @@ func TestAlertExecTemplate(t *testing.T) {
 		qFn := func(_ string) ([]datasource.Metric, error) {
 			return []datasource.Metric{
 				{
-					Labels: []prompbmarshal.Label{
+					Labels: []prompb.Label{
 						{Name: "foo", Value: "bar"},
 						{Name: "baz", Value: "qux"},
 					},
@@ -41,7 +41,7 @@ func TestAlertExecTemplate(t *testing.T) {
 					Timestamps: []int64{1},
 				},
 				{
-					Labels: []prompbmarshal.Label{
+					Labels: []prompb.Label{
 						{Name: "foo", Value: "garply"},
 						{Name: "baz", Value: "fred"},
 					},
@@ -213,7 +213,7 @@ func TestAlertExecTemplate(t *testing.T) {
 }
 
 func TestAlert_toPromLabels(t *testing.T) {
-	fn := func(labels map[string]string, exp []prompbmarshal.Label, relabel *promrelabel.ParsedConfigs) {
+	fn := func(labels map[string]string, exp []prompb.Label, relabel *promrelabel.ParsedConfigs) {
 		t.Helper()
 		a := Alert{Labels: labels}
 		got := a.applyRelabelingIfNeeded(relabel)
@@ -226,12 +226,12 @@ func TestAlert_toPromLabels(t *testing.T) {
 	fn(nil, nil, nil)
 	fn(
 		map[string]string{"foo": "bar", "a": "baz"}, // unsorted
-		[]prompbmarshal.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "bar"}},
+		[]prompb.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "bar"}},
 		nil,
 	)
 	fn(
 		map[string]string{"foo.bar": "baz", "service!name": "qux"},
-		[]prompbmarshal.Label{{Name: "foo_bar", Value: "baz"}, {Name: "service_name", Value: "qux"}},
+		[]prompb.Label{{Name: "foo_bar", Value: "baz"}, {Name: "service_name", Value: "qux"}},
 		nil,
 	)
 
@@ -247,17 +247,17 @@ func TestAlert_toPromLabels(t *testing.T) {
 
 	fn(
 		map[string]string{"a": "baz"},
-		[]prompbmarshal.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "aaa"}},
+		[]prompb.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "aaa"}},
 		pcs,
 	)
 	fn(
 		map[string]string{"foo": "bar", "a": "baz"},
-		[]prompbmarshal.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "aaa"}},
+		[]prompb.Label{{Name: "a", Value: "baz"}, {Name: "foo", Value: "aaa"}},
 		pcs,
 	)
 	fn(
 		map[string]string{"qux": "bar", "env": "prod", "environment": "production"},
-		[]prompbmarshal.Label{{Name: "foo", Value: "aaa"}, {Name: "qux", Value: "bar"}},
+		[]prompb.Label{{Name: "foo", Value: "aaa"}, {Name: "qux", Value: "bar"}},
 		pcs,
 	)
 }

app/vmalert/notifier/alertmanager.go

@@ -15,7 +15,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -86,7 +86,7 @@ func (am *AlertManager) Send(ctx context.Context, alerts []Alert, headers map[st
 func (am *AlertManager) send(ctx context.Context, alerts []Alert, headers map[string]string) error {
 	b := &bytes.Buffer{}
 	alertsToSend := make([]Alert, 0, len(alerts))
-	lblss := make([][]prompbmarshal.Label, 0, len(alerts))
+	lblss := make([][]prompb.Label, 0, len(alerts))
 	for _, a := range alerts {
 		lbls := a.applyRelabelingIfNeeded(am.relabelConfigs)
 		if len(lbls) == 0 {

app/vmalert/notifier/alertmanager_request.qtpl

@@ -1,11 +1,11 @@
 {% import (
     "time"
 
-    "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+    "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 ) %}
 {% stripspace %}
 
-{% func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) %}
+{% func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompb.Label) %}
 [
 {% for i, alert := range alerts %}
 {% code lbls := lblss[i] %}

app/vmalert/notifier/alertmanager_request.qtpl.go

@@ -8,7 +8,7 @@ package notifier
 import (
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
@@ -25,7 +25,7 @@ var (
 )
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
-func streamamRequest(qw422016 *qt422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) {
+func streamamRequest(qw422016 *qt422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompb.Label) {
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
 	qw422016.N().S(`[`)
 //line app/vmalert/notifier/alertmanager_request.qtpl:10
@@ -114,7 +114,7 @@ func streamamRequest(qw422016 *qt422016.Writer, alerts []Alert, generatorURL fun
 }
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:35
-func writeamRequest(qq422016 qtio422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) {
+func writeamRequest(qq422016 qtio422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompb.Label) {
 //line app/vmalert/notifier/alertmanager_request.qtpl:35
 	qw422016 := qt422016.AcquireWriter(qq422016)
 //line app/vmalert/notifier/alertmanager_request.qtpl:35
@@ -125,7 +125,7 @@ func writeamRequest(qq422016 qtio422016.Writer, alerts []Alert, generatorURL fun
 }
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:35
-func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) string {
+func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompb.Label) string {
 //line app/vmalert/notifier/alertmanager_request.qtpl:35
 	qb422016 := qt422016.AcquireByteBuffer()
 //line app/vmalert/notifier/alertmanager_request.qtpl:35

app/vmalert/notifier/config_watcher.go

@@ -248,7 +248,7 @@ func (cw *configWatcher) updateTargets(key TargetType, targetMetadata map[string
 	for _, ot := range oldTargets {
 		if _, ok := targetMetadata[ot.Addr()]; !ok {
 			// if target not exists in currentTargets, close it
-			ot.Notifier.Close()
+			ot.Close()
 		} else {
 			updatedTargets = append(updatedTargets, ot)
 			delete(targetMetadata, ot.Addr())

app/vmalert/notifier/config_watcher_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/consul"
 )
@@ -19,9 +20,9 @@ func TestConfigWatcherReload(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(f.Name()) }()
+	defer fs.MustRemovePath(f.Name())
 
-	writeToFile(t, f.Name(), `
+	writeToFile(f.Name(), `
 static_configs:
   - targets:
       - localhost:9093
@@ -41,9 +42,9 @@ static_configs:
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(f2.Name()) }()
+	defer fs.MustRemovePath(f2.Name())
 
-	writeToFile(t, f2.Name(), `
+	writeToFile(f2.Name(), `
 static_configs:
   - targets:
       - 127.0.0.1:9093
@@ -73,9 +74,9 @@ func TestConfigWatcherStart(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(consulSDFile.Name()) }()
+	defer fs.MustRemovePath(consulSDFile.Name())
 
-	writeToFile(t, consulSDFile.Name(), fmt.Sprintf(`
+	writeToFile(consulSDFile.Name(), fmt.Sprintf(`
 scheme: https
 path_prefix: proxy
 consul_sd_configs:
@@ -124,9 +125,9 @@ func TestConfigWatcherReloadConcurrent(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(consulSDFile.Name()) }()
+	defer fs.MustRemovePath(consulSDFile.Name())
 
-	writeToFile(t, consulSDFile.Name(), fmt.Sprintf(`
+	writeToFile(consulSDFile.Name(), fmt.Sprintf(`
 consul_sd_configs:
   - server: %s
     services:
@@ -140,9 +141,9 @@ consul_sd_configs:
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() { _ = os.Remove(staticAndConsulSDFile.Name()) }()
+	defer fs.MustRemovePath(staticAndConsulSDFile.Name())
 
-	writeToFile(t, staticAndConsulSDFile.Name(), fmt.Sprintf(`
+	writeToFile(staticAndConsulSDFile.Name(), fmt.Sprintf(`
 static_configs:
   - targets:
       - localhost:9093
@@ -187,9 +188,8 @@ consul_sd_configs:
 	wg.Wait()
 }
 
-func writeToFile(t *testing.T, file, b string) {
-	t.Helper()
-	checkErr(t, os.WriteFile(file, []byte(b), 0644))
+func writeToFile(file, b string) {
+	fs.MustWriteSync(file, []byte(b))
 }
 
 func checkErr(t *testing.T, err error) {

app/vmalert/remotewrite/client.go

@@ -20,7 +20,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -46,7 +46,7 @@ type Client struct {
 	addr          string
 	c             *http.Client
 	authCfg       *promauth.Config
-	input         chan prompbmarshal.TimeSeries
+	input         chan prompb.TimeSeries
 	flushInterval time.Duration
 	maxBatchSize  int
 	maxQueueSize  int
@@ -110,7 +110,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 		maxBatchSize:  cfg.MaxBatchSize,
 		maxQueueSize:  cfg.MaxQueueSize,
 		doneCh:        make(chan struct{}),
-		input:         make(chan prompbmarshal.TimeSeries, cfg.MaxQueueSize),
+		input:         make(chan prompb.TimeSeries, cfg.MaxQueueSize),
 	}
 
 	for i := 0; i < cc; i++ {
@@ -121,7 +121,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 
 // Push adds timeseries into queue for writing into remote storage.
 // Push returns and error if client is stopped or if queue is full.
-func (c *Client) Push(s prompbmarshal.TimeSeries) error {
+func (c *Client) Push(s prompb.TimeSeries) error {
 	rwTotal.Inc()
 	select {
 	case <-c.doneCh:
@@ -158,7 +158,7 @@ func (c *Client) Close() error {
 
 func (c *Client) run(ctx context.Context) {
 	ticker := time.NewTicker(c.flushInterval)
-	wr := &prompbmarshal.WriteRequest{}
+	wr := &prompb.WriteRequest{}
 	shutdown := func() {
 		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
 
@@ -221,7 +221,7 @@ func GetDroppedRows() int { return int(droppedRows.Get()) }
 // flush is a blocking function that marshals WriteRequest and sends
 // it to remote-write endpoint. Flush performs limited amount of retries
 // if request fails.
-func (c *Client) flush(ctx context.Context, wr *prompbmarshal.WriteRequest) {
+func (c *Client) flush(ctx context.Context, wr *prompb.WriteRequest) {
 	if len(wr.Timeseries) < 1 {
 		return
 	}

app/vmalert/remotewrite/client_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/golang/snappy"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
 func TestClient_Push(t *testing.T) {
@@ -46,8 +45,8 @@ func TestClient_Push(t *testing.T) {
 	r := rand.New(rand.NewSource(1))
 	const rowsN = int(1e4)
 	for i := 0; i < rowsN; i++ {
-		s := prompbmarshal.TimeSeries{
-			Samples: []prompbmarshal.Sample{{
+		s := prompb.TimeSeries{
+			Samples: []prompb.Sample{{
 				Value:     r.Float64(),
 				Timestamp: time.Now().Unix(),
 			}},
@@ -104,7 +103,7 @@ func TestClient_run_maxBatchSizeDuringShutdown(t *testing.T) {
 
 		// push time series to the client.
 		for i := 0; i < pushCnt; i++ {
-			if err = rwClient.Push(prompbmarshal.TimeSeries{}); err != nil {
+			if err = rwClient.Push(prompb.TimeSeries{}); err != nil {
 				t.Fatalf("cannot time series to the client: %s", err)
 			}
 		}
@@ -183,8 +182,9 @@ func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
 		rw.err(w, fmt.Errorf("decode err: %w", err))
 		return
 	}
-	wr := &prompb.WriteRequest{}
-	if err := wr.UnmarshalProtobuf(b); err != nil {
+	wru := &prompb.WriteRequestUnmarshaler{}
+	wr, err := wru.UnmarshalProtobuf(b)
+	if err != nil {
 		rw.err(w, fmt.Errorf("unmarhsal err: %w", err))
 		return
 	}

app/vmalert/remotewrite/debug_client.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 // DebugClient won't push series periodically, but will write data to remote endpoint
@@ -48,10 +48,10 @@ func NewDebugClient() (*DebugClient, error) {
 }
 
 // Push sends the given timeseries to the remote storage.
-func (c *DebugClient) Push(s prompbmarshal.TimeSeries) error {
+func (c *DebugClient) Push(s prompb.TimeSeries) error {
 	c.wg.Add(1)
 	defer c.wg.Done()
-	wr := &prompbmarshal.WriteRequest{Timeseries: []prompbmarshal.TimeSeries{s}}
+	wr := &prompb.WriteRequest{Timeseries: []prompb.TimeSeries{s}}
 	data := wr.MarshalProtobuf(nil)
 
 	return c.send(data)

app/vmalert/remotewrite/debug_client_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 func TestDebugClient_Push(t *testing.T) {
@@ -23,8 +23,8 @@ func TestDebugClient_Push(t *testing.T) {
 	const rowsN = 100
 	var sent int
 	for i := 0; i < rowsN; i++ {
-		s := prompbmarshal.TimeSeries{
-			Samples: []prompbmarshal.Sample{{
+		s := prompb.TimeSeries{
+			Samples: []prompb.Sample{{
 				Value:     float64(i),
 				Timestamp: time.Now().Unix(),
 			}},

app/vmalert/remotewrite/remotewrite.go

@@ -1,13 +1,13 @@
 package remotewrite
 
 import (
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 // RWClient represents an HTTP client for pushing data via remote write protocol
 type RWClient interface {
 	// Push pushes the give time series to remote storage
-	Push(s prompbmarshal.TimeSeries) error
+	Push(s prompb.TimeSeries) error
 	// Close stops the client. Client can't be reused after Close call.
 	Close() error
 }

app/vmalert/replay.go

@@ -28,8 +28,8 @@ var (
 		"Defines how many retries to make before giving up on rule if request for it returns an error.")
 	disableProgressBar = flag.Bool("replay.disableProgressBar", false, "Whether to disable rendering progress bars during the replay. "+
 		"Progress bar rendering might be verbose or break the logs parsing, so it is recommended to be disabled when not used in interactive mode.")
-	ruleEvaluationConcurrency = flag.Int("replay.ruleEvaluationConcurrency", 1, "The maximum number of concurrent `/query_range` requests for a single rule. "+
-		"Increasing this value when replaying for a long time and a single request range is limited by `-replay.maxDatapointsPerQuery`.")
+	ruleEvaluationConcurrency = flag.Int("replay.ruleEvaluationConcurrency", 1, "The maximum number of concurrent '/query_range' requests when replay recording rule or alerting rule with for=0. "+
+		"Increasing this value when replaying for a long time, since each request is limited by -replay.maxDatapointsPerQuery.")
 )
 
 func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw remotewrite.RWClient) (totalRows, droppedRows int, err error) {

app/vmalert/replay_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
@@ -23,7 +23,7 @@ func (fr *fakeReplayQuerier) BuildWithParams(_ datasource.QuerierParams) datasou
 
 type fakeRWClient struct{}
 
-func (fc *fakeRWClient) Push(_ prompbmarshal.TimeSeries) error {
+func (fc *fakeRWClient) Push(_ prompb.TimeSeries) error {
 	return nil
 }
 
@@ -246,24 +246,33 @@ func TestReplay(t *testing.T) {
 
 	// multiple rules + rule concurrency + group concurrency
 	f("2021-01-01T12:00:00.000Z", "2021-01-01T12:02:30.000Z", 1, 3, 0, []config.Group{
-		{Rules: []config.Rule{{Alert: "foo-group-single-concurrent", Expr: "sum(up) > 1"}, {Alert: "bar-group-single-concurrent", Expr: "max(up) < 1"}}, Concurrency: 2}}, &fakeReplayQuerier{
+		{Rules: []config.Rule{{Alert: "foo-group-single-concurrent", For: promutil.NewDuration(30 * time.Second), Expr: "sum(up) > 1"}, {Alert: "bar-group-single-concurrent", Expr: "max(up) < 1"}}, Concurrency: 2}}, &fakeReplayQuerier{
 		registry: map[string]map[string][]datasource.Metric{
 			"sum(up) > 1": {
-				"12:00:00+12:01:00": {},
-				"12:01:00+12:02:00": {{
-					Timestamps: []int64{1},
+				"12:00:00+12:01:00": {{
+					Timestamps: []int64{1609502460},
+					Values:     []float64{1},
+				}},
+				"12:01:00+12:02:00": {{
+					Timestamps: []int64{1609502520},
+					Values:     []float64{1},
+				}},
+				"12:02:00+12:02:30": {{
+					Timestamps: []int64{1609502580},
 					Values:     []float64{1},
 				}},
-				"12:02:00+12:02:30": {},
 			},
 			"max(up) < 1": {
-				"12:00:00+12:01:00": {},
+				"12:00:00+12:01:00": {{
+					Timestamps: []int64{1609502460},
+					Values:     []float64{1},
+				}},
 				"12:01:00+12:02:00": {{
-					Timestamps: []int64{1},
+					Timestamps: []int64{1609502520},
 					Values:     []float64{1},
 				}},
 				"12:02:00+12:02:30": {},
 			},
 		},
-	}, 4)
+	}, 10)
 }

app/vmalert/rule/alerting.go

@@ -19,7 +19,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -328,12 +328,12 @@ func (ar *AlertingRule) toLabels(m datasource.Metric, qFn templates.QueryFn) (*l
 // as this function modifies AlertingRule alerts state.
 // It is not thread safe.
 // It returns ALERT and ALERT_FOR_STATE time series as a result.
-func (ar *AlertingRule) execRange(ctx context.Context, start, end time.Time) ([]prompbmarshal.TimeSeries, error) {
+func (ar *AlertingRule) execRange(ctx context.Context, start, end time.Time) ([]prompb.TimeSeries, error) {
 	res, err := ar.q.QueryRange(ctx, ar.Expr, start, end)
 	if err != nil {
 		return nil, err
 	}
-	var result []prompbmarshal.TimeSeries
+	var result []prompb.TimeSeries
 	holdAlertState := make(map[uint64]*notifier.Alert)
 	qFn := func(_ string) ([]datasource.Metric, error) {
 		logger.Warnf("`query` template isn't supported in replay mode, mocked data is used")
@@ -341,11 +341,15 @@ func (ar *AlertingRule) execRange(ctx context.Context, start, end time.Time) ([]
 		return []datasource.Metric{{Timestamps: []int64{0}, Values: []float64{math.NaN()}}}, nil
 	}
 	for _, s := range res.Data {
-		ls, as, err := ar.expandTemplates(s, qFn, time.Time{})
+		ls, err := ar.expandLabelTemplates(s)
 		if err != nil {
-			return nil, fmt.Errorf("failed to expand templates: %s", err)
+			return nil, err
 		}
 		alertID := hash(ls.processed)
+		as, err := ar.expandAnnotationTemplates(s, qFn, time.Time{}, ls)
+		if err != nil {
+			return nil, err
+		}
 		a := ar.newAlert(s, time.Time{}, ls.processed, as) // initial alert
 
 		prevT := time.Time{}
@@ -363,7 +367,7 @@ func (ar *AlertingRule) execRange(ctx context.Context, start, end time.Time) ([]
 				a.State = notifier.StatePending
 				a.ActiveAt = at
 				// re-template the annotations as active timestamp is changed
-				_, a.Annotations, _ = ar.expandTemplates(s, qFn, at)
+				a.Annotations, _ = ar.expandAnnotationTemplates(s, qFn, at, ls)
 				a.Start = time.Time{}
 			} else if at.Sub(a.ActiveAt) >= ar.For && a.State != notifier.StateFiring {
 				a.State = notifier.StateFiring
@@ -376,13 +380,15 @@ func (ar *AlertingRule) execRange(ctx context.Context, start, end time.Time) ([]
 			}
 			result = append(result, ar.alertToTimeSeries(a, s.Timestamps[i])...)
 
-			// save alert's state on last iteration, so it can be used on the next execRange call
-			if at.Equal(end) {
+			// if for>0, save alert's state on last iteration, so it can be used on the next execRange call
+			if ar.For > 0 && at.Equal(end) {
 				holdAlertState[alertID] = a
 			}
 		}
 	}
-	ar.alerts = holdAlertState
+	if len(holdAlertState) > 0 {
+		ar.alerts = holdAlertState
+	}
 	return result, nil
 }
 
@@ -392,7 +398,7 @@ const resolvedRetention = 15 * time.Minute
 
 // exec executes AlertingRule expression via the given Querier.
 // Based on the Querier results AlertingRule maintains notifier.Alerts
-func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompbmarshal.TimeSeries, error) {
+func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := ar.q.Query(ctx, ar.Expr, ts)
 	curState := StateEntry{
@@ -428,9 +434,22 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 	expandedLabels := make([]*labelSet, len(res.Data))
 	expandedAnnotations := make([]map[string]string, len(res.Data))
 	for i, m := range res.Data {
-		ls, as, err := ar.expandTemplates(m, qFn, ts)
+		ls, err := ar.expandLabelTemplates(m)
 		if err != nil {
-			curState.Err = fmt.Errorf("failed to expand templates: %w", err)
+			curState.Err = err
+			return nil, curState.Err
+		}
+		at := ts
+		alertID := hash(ls.processed)
+		if a, ok := ar.alerts[alertID]; ok {
+			// modify activeAt for annotation templating if the alert has already triggered(in state Pending or Firing)
+			if a.State != notifier.StateInactive {
+				at = a.ActiveAt
+			}
+		}
+		as, err := ar.expandAnnotationTemplates(m, qFn, at, ls)
+		if err != nil {
+			curState.Err = err
 			return nil, curState.Err
 		}
 		expandedLabels[i] = ls
@@ -473,6 +492,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 			a.KeepFiringSince = time.Time{}
 			continue
 		}
+
 		a := ar.newAlert(m, ts, labels.processed, annotations)
 		a.ID = alertID
 		a.State = notifier.StatePending
@@ -480,7 +500,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 		ar.logDebugf(ts, a, "created in state PENDING")
 	}
 	var numActivePending int
-	var tss []prompbmarshal.TimeSeries
+	var tss []prompb.TimeSeries
 	for h, a := range ar.alerts {
 		// if alert wasn't updated in this iteration
 		// means it is resolved already
@@ -536,12 +556,18 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 	return append(tss, ar.toTimeSeries(ts.Unix())...), nil
 }
 
-func (ar *AlertingRule) expandTemplates(m datasource.Metric, qFn templates.QueryFn, ts time.Time) (*labelSet, map[string]string, error) {
+func (ar *AlertingRule) expandLabelTemplates(m datasource.Metric) (*labelSet, error) {
+	qFn := func(_ string) ([]datasource.Metric, error) {
+		return nil, fmt.Errorf("`query` template isn't supported in rule label")
+	}
 	ls, err := ar.toLabels(m, qFn)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to expand labels: %w", err)
+		return nil, fmt.Errorf("failed to expand label templates: %s", err)
 	}
+	return ls, nil
+}
 
+func (ar *AlertingRule) expandAnnotationTemplates(m datasource.Metric, qFn templates.QueryFn, activeAt time.Time, ls *labelSet) (map[string]string, error) {
 	tplData := notifier.AlertTplData{
 		Value:    m.Values[0],
 		Type:     ar.Type.String(),
@@ -549,19 +575,19 @@ func (ar *AlertingRule) expandTemplates(m datasource.Metric, qFn templates.Query
 		Expr:     ar.Expr,
 		AlertID:  hash(ls.processed),
 		GroupID:  ar.GroupID,
-		ActiveAt: ts,
+		ActiveAt: activeAt,
 		For:      ar.For,
 	}
 	as, err := notifier.ExecTemplate(qFn, ar.Annotations, tplData)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to template annotations: %w", err)
+		return nil, fmt.Errorf("failed to expand annotation templates: %s", err)
 	}
-	return ls, as, nil
+	return as, nil
 }
 
 // toTimeSeries creates `ALERTS` and `ALERTS_FOR_STATE` for active alerts
-func (ar *AlertingRule) toTimeSeries(timestamp int64) []prompbmarshal.TimeSeries {
-	var tss []prompbmarshal.TimeSeries
+func (ar *AlertingRule) toTimeSeries(timestamp int64) []prompb.TimeSeries {
+	var tss []prompb.TimeSeries
 	for _, a := range ar.alerts {
 		if a.State == notifier.StateInactive {
 			continue
@@ -632,67 +658,67 @@ const (
 )
 
 // alertToTimeSeries converts the given alert with the given timestamp to time series
-func (ar *AlertingRule) alertToTimeSeries(a *notifier.Alert, timestamp int64) []prompbmarshal.TimeSeries {
-	return []prompbmarshal.TimeSeries{
+func (ar *AlertingRule) alertToTimeSeries(a *notifier.Alert, timestamp int64) []prompb.TimeSeries {
+	return []prompb.TimeSeries{
 		alertToTimeSeries(a, timestamp),
 		alertForToTimeSeries(a, timestamp),
 	}
 }
 
-func alertToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	labels := make([]prompbmarshal.Label, 0, len(a.Labels)+2)
+func alertToTimeSeries(a *notifier.Alert, timestamp int64) prompb.TimeSeries {
+	labels := make([]prompb.Label, 0, len(a.Labels)+2)
 	for k, v := range a.Labels {
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
 	// __name__ already been dropped, no need to check duplication
-	labels = append(labels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
+	labels = append(labels, prompb.Label{Name: "__name__", Value: alertMetricName})
 	if ol := promrelabel.GetLabelByName(labels, alertStateLabel); ol != nil {
 		ol.Value = a.State.String()
 	} else {
-		labels = append(labels, prompbmarshal.Label{Name: alertStateLabel, Value: a.State.String()})
+		labels = append(labels, prompb.Label{Name: alertStateLabel, Value: a.State.String()})
 	}
 	return newTimeSeries([]float64{1}, []int64{timestamp}, labels)
 }
 
 // alertForToTimeSeries returns a time series that represents
 // state of active alerts, where value is time when alert become active
-func alertForToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	labels := make([]prompbmarshal.Label, 0, len(a.Labels)+1)
+func alertForToTimeSeries(a *notifier.Alert, timestamp int64) prompb.TimeSeries {
+	labels := make([]prompb.Label, 0, len(a.Labels)+1)
 	for k, v := range a.Labels {
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
 	// __name__ already been dropped, no need to check duplication
-	labels = append(labels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+	labels = append(labels, prompb.Label{Name: "__name__", Value: alertForStateMetricName})
 	return newTimeSeries([]float64{float64(a.ActiveAt.Unix())}, []int64{timestamp}, labels)
 }
 
 // pendingAlertStaleTimeSeries returns stale `ALERTS` and `ALERTS_FOR_STATE` time series
 // for alerts which changed their state from Pending to Inactive or Firing.
-func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeAlertForState bool) []prompbmarshal.TimeSeries {
-	var result []prompbmarshal.TimeSeries
-	baseLabels := make([]prompbmarshal.Label, 0, len(ls)+1)
+func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeAlertForState bool) []prompb.TimeSeries {
+	var result []prompb.TimeSeries
+	baseLabels := make([]prompb.Label, 0, len(ls)+1)
 	for k, v := range ls {
-		baseLabels = append(baseLabels, prompbmarshal.Label{
+		baseLabels = append(baseLabels, prompb.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
 
-	alertsLabels := make([]prompbmarshal.Label, 0, len(ls)+2)
+	alertsLabels := make([]prompb.Label, 0, len(ls)+2)
 	alertsLabels = append(alertsLabels, baseLabels...)
 	// __name__ already been dropped, no need to check duplication
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StatePending.String()})
+	alertsLabels = append(alertsLabels, prompb.Label{Name: "__name__", Value: alertMetricName})
+	alertsLabels = append(alertsLabels, prompb.Label{Name: alertStateLabel, Value: notifier.StatePending.String()})
 	result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels))
 
 	if includeAlertForState {
-		baseLabels = append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+		baseLabels = append(baseLabels, prompb.Label{Name: "__name__", Value: alertForStateMetricName})
 		result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, baseLabels))
 	}
 	return result
@@ -700,24 +726,24 @@ func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeA
 
 // firingAlertStaleTimeSeries returns stale `ALERTS` and `ALERTS_FOR_STATE` time series
 // for alerts which changed their state from Firing to Inactive.
-func firingAlertStaleTimeSeries(ls map[string]string, timestamp int64) []prompbmarshal.TimeSeries {
-	baseLabels := make([]prompbmarshal.Label, 0, len(ls)+1)
+func firingAlertStaleTimeSeries(ls map[string]string, timestamp int64) []prompb.TimeSeries {
+	baseLabels := make([]prompb.Label, 0, len(ls)+1)
 	for k, v := range ls {
-		baseLabels = append(baseLabels, prompbmarshal.Label{
+		baseLabels = append(baseLabels, prompb.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
 
-	alertsLabels := make([]prompbmarshal.Label, 0, len(ls)+2)
+	alertsLabels := make([]prompb.Label, 0, len(ls)+2)
 	alertsLabels = append(alertsLabels, baseLabels...)
 	// __name__ already been dropped, no need to check duplication
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StateFiring.String()})
+	alertsLabels = append(alertsLabels, prompb.Label{Name: "__name__", Value: alertMetricName})
+	alertsLabels = append(alertsLabels, prompb.Label{Name: alertStateLabel, Value: notifier.StateFiring.String()})
 
-	baseLabels = append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+	baseLabels = append(baseLabels, prompb.Label{Name: "__name__", Value: alertForStateMetricName})
 
-	return []prompbmarshal.TimeSeries{
+	return []prompb.TimeSeries{
 		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels),
 		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, baseLabels),
 	}

app/vmalert/rule/alerting_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -18,7 +19,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
@@ -51,7 +52,7 @@ func TestNewAlertingRule(t *testing.T) {
 func TestAlertingRuleToTimeSeries(t *testing.T) {
 	timestamp := time.Now()
 
-	f := func(rule *AlertingRule, alert *notifier.Alert, tssExpected []prompbmarshal.TimeSeries) {
+	f := func(rule *AlertingRule, alert *notifier.Alert, tssExpected []prompb.TimeSeries) {
 		t.Helper()
 
 		rule.alerts[alert.ID] = alert
@@ -64,8 +65,8 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 	f(newTestAlertingRule("instant", 0), &notifier.Alert{
 		State:    notifier.StateFiring,
 		ActiveAt: timestamp.Add(time.Second),
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: alertMetricName,
@@ -77,7 +78,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: alertForStateMetricName,
@@ -91,9 +92,9 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 			"job":      "foo",
 			"instance": "bar",
 		},
-	}, []prompbmarshal.TimeSeries{
+	}, []prompb.TimeSeries{
 		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: alertMetricName,
@@ -113,7 +114,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 			}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: alertForStateMetricName,
@@ -134,8 +135,8 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		Labels: map[string]string{
 			alertStateLabel: "foo",
 		},
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: alertMetricName,
@@ -147,7 +148,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: alertForStateMetricName,
@@ -162,8 +163,8 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 	f(newTestAlertingRule("for", time.Second), &notifier.Alert{
 		State:    notifier.StateFiring,
 		ActiveAt: timestamp.Add(time.Second),
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: alertMetricName,
@@ -175,7 +176,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: alertForStateMetricName,
@@ -186,8 +187,8 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 	f(newTestAlertingRule("for pending", 10*time.Second), &notifier.Alert{
 		State:    notifier.StatePending,
 		ActiveAt: timestamp.Add(time.Second),
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: alertMetricName,
@@ -197,7 +198,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 				Value: notifier.StatePending.String(),
 			},
 		}),
-		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: alertForStateMetricName,
@@ -215,7 +216,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 
 	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
 
-	f := func(rule *AlertingRule, steps [][]datasource.Metric, alertsExpected map[int][]testAlert, tssExpected map[int][]prompbmarshal.TimeSeries) {
+	f := func(rule *AlertingRule, steps [][]datasource.Metric, alertsExpected map[int][]testAlert, tssExpected map[int][]prompb.TimeSeries) {
 		t.Helper()
 
 		fq := &datasource.FakeQuerier{}
@@ -267,8 +268,15 @@ func TestAlertingRule_Exec(t *testing.T) {
 				if got.State != exp.State {
 					t.Fatalf("evalIndex %d: expected state %d; got %d", i, exp.State, got.State)
 				}
+				if rule.Annotations != nil && exp.Annotations != nil {
+					if !reflect.DeepEqual(got.Annotations, exp.Annotations) {
+						t.Fatalf("evalIndex %d: expected annotations %v; got %v", i, exp.Annotations, got.Annotations)
+					}
+				}
 			}
 		}
+		// reset ts for next test
+		ts, _ = time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
 	}
 
 	f(newTestAlertingRule("empty", 0), [][]datasource.Metric{}, nil, nil)
@@ -278,15 +286,15 @@ func TestAlertingRule_Exec(t *testing.T) {
 	}, map[int][]testAlert{
 		0: {{alert: &notifier.Alert{State: notifier.StateFiring}}},
 	},
-		map[int][]prompbmarshal.TimeSeries{
+		map[int][]prompb.TimeSeries{
 			0: {
 				{
-					Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "empty_labels"}, {Name: "alertstate", Value: "firing"}},
-					Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
+					Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "empty_labels"}, {Name: "alertstate", Value: "firing"}},
+					Samples: []prompb.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
 				},
 				{
-					Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "empty_labels"}},
-					Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
+					Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "empty_labels"}},
+					Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
 				},
 			},
 		})
@@ -303,35 +311,35 @@ func TestAlertingRule_Exec(t *testing.T) {
 		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
 		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
-	}, map[int][]prompbmarshal.TimeSeries{
+	}, map[int][]prompb.TimeSeries{
 		0: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
 			},
 		},
 		1: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 		},
 		2: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 		},
 	})
@@ -386,55 +394,55 @@ func TestAlertingRule_Exec(t *testing.T) {
 			{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateInactive}},
 			{labels: []string{"name", "foo2"}, alert: &notifier.Alert{State: notifier.StateFiring}},
 		},
-	}, map[int][]prompbmarshal.TimeSeries{
+	}, map[int][]prompb.TimeSeries{
 		0: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
 			},
 		},
 		1: {
 			// stale time series for foo, `firing -> inactive`
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			// new time series for foo1
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 		},
 		2: {
 			// stale time series for foo1
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 			// new time series for foo2
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo2"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo2"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 		},
 	})
@@ -451,30 +459,30 @@ func TestAlertingRule_Exec(t *testing.T) {
 	}, map[int][]testAlert{
 		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, map[int][]prompbmarshal.TimeSeries{
+	}, map[int][]prompb.TimeSeries{
 		0: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
 			},
 		},
 		1: {
 			// stale time series for `pending -> firing`
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 		},
 	})
@@ -488,41 +496,41 @@ func TestAlertingRule_Exec(t *testing.T) {
 		0: {{labels: []string{"name", "foo", "a1", "b1", "a2", "b2", "a3", "b3"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		1: {{labels: []string{"name", "foo", "a1", "b1", "a2", "b2", "a3", "b3"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		2: {},
-	}, map[int][]prompbmarshal.TimeSeries{
+	}, map[int][]prompb.TimeSeries{
 		0: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
 			},
 		},
 		1: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: float64(ts.Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
 			},
 		},
 		// stale time series for `pending -> inactive`
 		2: {
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
+				Labels:  []prompb.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompb.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
 			},
 		},
 	})
 
-	f(newTestAlertingRule("for-pending=>firing=>inactive=>pending=>firing", defaultStep), [][]datasource.Metric{
+	f(newTestAlertingRuleWithCustomFields("for-pending=>firing=>inactive=>pending=>firing", defaultStep, 0, 0, map[string]string{"activeAt": "{{ $activeAt.UnixMilli }}"}), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
 		{metricWithLabels(t, "name", "foo")},
 		// empty step to set alert inactive
@@ -530,11 +538,11 @@ func TestAlertingRule_Exec(t *testing.T) {
 		{metricWithLabels(t, "name", "foo")},
 		{metricWithLabels(t, "name", "foo")},
 	}, map[int][]testAlert{
-		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
-		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
-		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
-		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
+		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending, Annotations: map[string]string{"activeAt": strconv.FormatInt(ts.UnixMilli(), 10)}}}},
+		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring, Annotations: map[string]string{"activeAt": strconv.FormatInt(ts.UnixMilli(), 10)}}}},
+		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive, Annotations: map[string]string{"activeAt": strconv.FormatInt(ts.UnixMilli(), 10)}}}},
+		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending, Annotations: map[string]string{"activeAt": strconv.FormatInt(ts.Add(defaultStep*3).UnixMilli(), 10)}}}},
+		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring, Annotations: map[string]string{"activeAt": strconv.FormatInt(ts.Add(defaultStep*3).UnixMilli(), 10)}}}},
 	}, nil)
 
 	f(newTestAlertingRuleWithCustomFields("for-pending=>firing=>keepfiring=>firing", defaultStep, 0, defaultStep, nil), [][]datasource.Metric{
@@ -588,7 +596,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		var expTS []prompbmarshal.TimeSeries
+		var expTS []prompb.TimeSeries
 		var j int
 		for _, series := range data {
 			for _, timestamp := range series.Timestamps {
@@ -728,7 +736,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{Values: []float64{1, 1, 1}, Timestamps: []int64{1, 3, 5}},
 		{
 			Values: []float64{1, 1}, Timestamps: []int64{1, 5},
-			Labels: []prompbmarshal.Label{{Name: "foo", Value: "bar"}},
+			Labels: []prompb.Label{{Name: "foo", Value: "bar"}},
 		},
 	}, []*notifier.Alert{
 		{State: notifier.StatePending, ActiveAt: time.Unix(1, 0)},
@@ -776,7 +784,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{Values: []float64{1, 1}, Timestamps: []int64{1, 100}},
 		{
 			Values: []float64{1, 1}, Timestamps: []int64{1, 5},
-			Labels: []prompbmarshal.Label{{Name: "foo", Value: "bar"}},
+			Labels: []prompb.Label{{Name: "foo", Value: "bar"}},
 		},
 	}, []*notifier.Alert{
 		{
@@ -1353,7 +1361,7 @@ func newTestAlertingRuleWithCustomFields(name string, waitFor, evalInterval, kee
 
 func TestAlertingRule_ToLabels(t *testing.T) {
 	metric := datasource.Metric{
-		Labels: []prompbmarshal.Label{
+		Labels: []prompb.Label{
 			{Name: "instance", Value: "0.0.0.0:8800"},
 			{Name: "group", Value: "vmalert"},
 			{Name: "alertname", Value: "ConfigurationReloadFailure"},

app/vmalert/rule/group.go

@@ -21,7 +21,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 var (
@@ -587,6 +587,11 @@ func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoi
 
 func replayRuleRange(r Rule, ri rangeIterator, bar *pb.ProgressBar, rw remotewrite.RWClient, replayRuleRetryAttempts, ruleEvaluationConcurrency int) int {
 	fmt.Printf("> Rule %q (ID: %d)\n", r, r.ID())
+	// alerting rule with for>0 can't be replayed concurrently, since the status change might depend on the previous evaluation
+	// see https://github.com/VictoriaMetrics/VictoriaMetrics/commit/abcb21aa5ee918ba9a4e9cde495dba06e1e9564c
+	if r, ok := r.(*AlertingRule); ok && r.For > 0 {
+		ruleEvaluationConcurrency = 1
+	}
 	sem := make(chan struct{}, ruleEvaluationConcurrency)
 	wg := sync.WaitGroup{}
 	res := make(chan int, int(ri.end.Sub(ri.start)/ri.step)+1)
@@ -755,7 +760,7 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 	}
 
 	if e.Rw != nil {
-		pushToRW := func(tss []prompbmarshal.TimeSeries) error {
+		pushToRW := func(tss []prompb.TimeSeries) error {
 			var lastErr error
 			for _, ts := range tss {
 				if err := e.Rw.Push(ts); err != nil {

app/vmalert/rule/recording.go

@@ -14,7 +14,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -132,13 +132,13 @@ func (rr *RecordingRule) unregisterMetrics() {
 // execRange executes recording rule on the given time range similarly to Exec.
 // It doesn't update internal states of the Rule and meant to be used just
 // to get time series for backfilling.
-func (rr *RecordingRule) execRange(ctx context.Context, start, end time.Time) ([]prompbmarshal.TimeSeries, error) {
+func (rr *RecordingRule) execRange(ctx context.Context, start, end time.Time) ([]prompb.TimeSeries, error) {
 	res, err := rr.q.QueryRange(ctx, rr.Expr, start, end)
 	if err != nil {
 		return nil, err
 	}
 	duplicates := make(map[string]struct{}, len(res.Data))
-	var tss []prompbmarshal.TimeSeries
+	var tss []prompb.TimeSeries
 	for _, s := range res.Data {
 		ts := rr.toTimeSeries(s)
 		key := stringifyLabels(ts.Labels)
@@ -152,7 +152,7 @@ func (rr *RecordingRule) execRange(ctx context.Context, start, end time.Time) ([
 }
 
 // exec executes RecordingRule expression via the given Querier.
-func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompbmarshal.TimeSeries, error) {
+func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := rr.q.Query(ctx, rr.Expr, ts)
 	curState := StateEntry{
@@ -187,7 +187,7 @@ func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]p
 
 	curEvaluation := make(map[string]struct{}, len(qMetrics))
 	lastEvaluation := rr.lastEvaluation
-	var tss []prompbmarshal.TimeSeries
+	var tss []prompb.TimeSeries
 	for _, r := range qMetrics {
 		ts := rr.toTimeSeries(r)
 		key := stringifyLabels(ts.Labels)
@@ -201,9 +201,9 @@ func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]p
 	}
 	// check for stale time series
 	for k := range lastEvaluation {
-		tss = append(tss, prompbmarshal.TimeSeries{
+		tss = append(tss, prompb.TimeSeries{
 			Labels: stringToLabels(k),
-			Samples: []prompbmarshal.Sample{
+			Samples: []prompb.Sample{
 				{Value: decimal.StaleNaN, Timestamp: ts.UnixNano() / 1e6},
 			}})
 	}
@@ -222,12 +222,12 @@ func (rr *RecordingRule) logDebugf(at time.Time, format string, args ...any) {
 	logger.Infof("%s", prefix+msg)
 }
 
-func stringToLabels(s string) []prompbmarshal.Label {
+func stringToLabels(s string) []prompb.Label {
 	labels := strings.Split(s, ",")
-	rLabels := make([]prompbmarshal.Label, 0, len(labels))
+	rLabels := make([]prompb.Label, 0, len(labels))
 	for i := range labels {
 		if label := strings.Split(labels[i], "="); len(label) == 2 {
-			rLabels = append(rLabels, prompbmarshal.Label{
+			rLabels = append(rLabels, prompb.Label{
 				Name:  label[0],
 				Value: label[1],
 			})
@@ -236,7 +236,7 @@ func stringToLabels(s string) []prompbmarshal.Label {
 	return rLabels
 }
 
-func stringifyLabels(labels []prompbmarshal.Label) string {
+func stringifyLabels(labels []prompb.Label) string {
 	b := strings.Builder{}
 	for i, l := range labels {
 		b.WriteString(l.Name)
@@ -249,11 +249,11 @@ func stringifyLabels(labels []prompbmarshal.Label) string {
 	return b.String()
 }
 
-func (rr *RecordingRule) toTimeSeries(m datasource.Metric) prompbmarshal.TimeSeries {
+func (rr *RecordingRule) toTimeSeries(m datasource.Metric) prompb.TimeSeries {
 	if preN := promrelabel.GetLabelByName(m.Labels, "__name__"); preN != nil {
 		preN.Value = rr.Name
 	} else {
-		m.Labels = append(m.Labels, prompbmarshal.Label{
+		m.Labels = append(m.Labels, prompb.Label{
 			Name:  "__name__",
 			Value: rr.Name,
 		})
@@ -270,7 +270,7 @@ func (rr *RecordingRule) toTimeSeries(m datasource.Metric) prompbmarshal.TimeSer
 			existingLabel.Name = fmt.Sprintf("exported_%s", existingLabel.Name)
 		}
 		// add extra label
-		m.Labels = append(m.Labels, prompbmarshal.Label{
+		m.Labels = append(m.Labels, prompb.Label{
 			Name:  k,
 			Value: rr.Labels[k],
 		})

app/vmalert/rule/recording_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 func TestNewRecordingRule(t *testing.T) {
@@ -41,7 +41,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
 	const defaultStep = 5 * time.Millisecond
 
-	f := func(rule *RecordingRule, steps [][]datasource.Metric, tssExpected [][]prompbmarshal.TimeSeries) {
+	f := func(rule *RecordingRule, steps [][]datasource.Metric, tssExpected [][]prompb.TimeSeries) {
 		t.Helper()
 
 		fq := &datasource.FakeQuerier{}
@@ -68,8 +68,8 @@ func TestRecordingRule_Exec(t *testing.T) {
 		Name: "foo",
 	}, [][]datasource.Metric{{
 		metricWithValueAndLabels(t, 10, "__name__", "bar"),
-	}}, [][]prompbmarshal.TimeSeries{{
-		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+	}}, [][]prompb.TimeSeries{{
+		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "foo",
@@ -90,9 +90,9 @@ func TestRecordingRule_Exec(t *testing.T) {
 		{
 			metricWithValueAndLabels(t, 10, "__name__", "foo", "job", "bar"),
 		},
-	}, [][]prompbmarshal.TimeSeries{
+	}, [][]prompb.TimeSeries{
 		{
-			newTimeSeries([]float64{1}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{1}, []int64{ts.UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -102,7 +102,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 					Value: "foo",
 				},
 			}),
-			newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -114,7 +114,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 			}),
 		},
 		{
-			newTimeSeries([]float64{10}, []int64{ts.Add(defaultStep).UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{10}, []int64{ts.Add(defaultStep).UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -125,7 +125,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 				},
 			}),
 			// stale time series
-			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(defaultStep).UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(defaultStep).UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -137,7 +137,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 			}),
 		},
 		{
-			newTimeSeries([]float64{10}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{10}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -147,7 +147,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 					Value: "bar",
 				},
 			}),
-			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompbmarshal.Label{
+			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -169,8 +169,8 @@ func TestRecordingRule_Exec(t *testing.T) {
 		metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "foo"),
 		metricWithValueAndLabels(t, 1, "__name__", "bar", "job", "bar", "source", "origin"),
 		metricWithValueAndLabels(t, 1, "__name__", "baz", "job", "baz", "source", "test"),
-	}}, [][]prompbmarshal.TimeSeries{{
-		newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+	}}, [][]prompb.TimeSeries{{
+		newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "job:foo",
@@ -185,7 +185,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 			},
 		}),
 		newTimeSeries([]float64{1}, []int64{ts.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "job:foo",
@@ -204,7 +204,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 				},
 			}),
 		newTimeSeries([]float64{1}, []int64{ts.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "job:foo",
@@ -222,7 +222,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 }
 
 func TestRecordingRule_ExecRange(t *testing.T) {
-	f := func(rule *RecordingRule, metrics []datasource.Metric, tssExpected []prompbmarshal.TimeSeries) {
+	f := func(rule *RecordingRule, metrics []datasource.Metric, tssExpected []prompb.TimeSeries) {
 		t.Helper()
 
 		fq := &datasource.FakeQuerier{}
@@ -243,9 +243,9 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 		Name: "foo",
 	}, []datasource.Metric{
 		metricWithValuesAndLabels(t, []float64{10, 20, 30}, "__name__", "bar"),
-	}, []prompbmarshal.TimeSeries{
+	}, []prompb.TimeSeries{
 		newTimeSeries([]float64{10, 20, 30}, []int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foo",
@@ -259,8 +259,8 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 		metricWithValuesAndLabels(t, []float64{1}, "__name__", "foo", "job", "foo"),
 		metricWithValuesAndLabels(t, []float64{2, 3}, "__name__", "bar", "job", "bar"),
 		metricWithValuesAndLabels(t, []float64{4, 5, 6}, "__name__", "baz", "job", "baz"),
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "foobarbaz",
@@ -270,7 +270,7 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 				Value: "foo",
 			},
 		}),
-		newTimeSeries([]float64{2, 3}, []int64{timestamp.UnixNano(), timestamp.UnixNano()}, []prompbmarshal.Label{
+		newTimeSeries([]float64{2, 3}, []int64{timestamp.UnixNano(), timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "foobarbaz",
@@ -281,7 +281,7 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 			},
 		}),
 		newTimeSeries([]float64{4, 5, 6},
-			[]int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()}, []prompbmarshal.Label{
+			[]int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()}, []prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "foobarbaz",
@@ -301,8 +301,8 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 	}, []datasource.Metric{
 		metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "foo"),
 		metricWithValueAndLabels(t, 1, "__name__", "bar", "job", "bar"),
-	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
+	}, []prompb.TimeSeries{
+		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "job:foo",
@@ -317,7 +317,7 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 			},
 		}),
 		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
+			[]prompb.Label{
 				{
 					Name:  "__name__",
 					Value: "job:foo",
@@ -437,7 +437,7 @@ func TestRecordingRuleExec_Negative(t *testing.T) {
 
 	_, err = rr.exec(context.TODO(), time.Now(), 0)
 	if err != nil {
-		t.Fatalf("cannot execute recroding rule: %s", err)
+		t.Fatalf("cannot execute recording rule: %s", err)
 	}
 }
 
@@ -480,8 +480,8 @@ func TestRecordingRuleExec_Partial(t *testing.T) {
 	rule.Debug = true
 	rule.q = fq
 	got, err := rule.exec(context.TODO(), ts, 0)
-	want := []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+	want := []prompb.TimeSeries{
+		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompb.Label{
 			{
 				Name:  "__name__",
 				Value: "foo",

app/vmalert/rule/rule.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 // Rule represents alerting or recording rule
@@ -23,9 +23,9 @@ type Rule interface {
 	ID() uint64
 	// exec executes the rule with given context at the given timestamp and limit.
 	// returns an err if number of resulting time series exceeds the limit.
-	exec(ctx context.Context, ts time.Time, limit int) ([]prompbmarshal.TimeSeries, error)
+	exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error)
 	// execRange executes the rule on the given time range.
-	execRange(ctx context.Context, start, end time.Time) ([]prompbmarshal.TimeSeries, error)
+	execRange(ctx context.Context, start, end time.Time) ([]prompb.TimeSeries, error)
 	// updateWith performs modification of current Rule
 	// with fields of the given Rule.
 	updateWith(Rule) error
@@ -151,7 +151,7 @@ func (s *ruleState) add(e StateEntry) {
 
 func replayRule(r Rule, start, end time.Time, rw remotewrite.RWClient, replayRuleRetryAttempts int) (int, error) {
 	var err error
-	var tss []prompbmarshal.TimeSeries
+	var tss []prompb.TimeSeries
 	for i := 0; i < replayRuleRetryAttempts; i++ {
 		tss, err = r.execRange(context.Background(), start, end)
 		if err == nil {

app/vmalert/rule/test_helpers.go

@@ -9,7 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 // CompareRules is a test helper func for other tests
@@ -94,7 +94,7 @@ func metricWithLabels(t *testing.T, labels ...string) datasource.Metric {
 	}
 	m := datasource.Metric{Values: []float64{1}, Timestamps: []int64{1}}
 	for i := 0; i < len(labels); i += 2 {
-		m.Labels = append(m.Labels, prompbmarshal.Label{
+		m.Labels = append(m.Labels, prompb.Label{
 			Name:  labels[i],
 			Value: labels[i+1],
 		})
@@ -102,7 +102,7 @@ func metricWithLabels(t *testing.T, labels ...string) datasource.Metric {
 	return m
 }
 
-func compareTimeSeries(t *testing.T, a, b []prompbmarshal.TimeSeries) error {
+func compareTimeSeries(t *testing.T, a, b []prompb.TimeSeries) error {
 	t.Helper()
 	if len(a) != len(b) {
 		return fmt.Errorf("expected number of timeseries %d; got %d", len(a), len(b))

app/vmalert/rule/utils.go

@@ -8,19 +8,19 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
 // newTimeSeries first sorts given labels, then returns new time series.
-func newTimeSeries(values []float64, timestamps []int64, labels []prompbmarshal.Label) prompbmarshal.TimeSeries {
+func newTimeSeries(values []float64, timestamps []int64, labels []prompb.Label) prompb.TimeSeries {
 	promrelabel.SortLabels(labels)
-	ts := prompbmarshal.TimeSeries{
+	ts := prompb.TimeSeries{
 		Labels:  labels,
-		Samples: make([]prompbmarshal.Sample, len(values)),
+		Samples: make([]prompb.Sample, len(values)),
 	}
 	for i := range values {
-		ts.Samples[i] = prompbmarshal.Sample{
+		ts.Samples[i] = prompb.Sample{
 			Value:     values[i],
 			Timestamp: time.Unix(timestamps[i], 0).UnixNano() / 1e6,
 		}
@@ -39,7 +39,7 @@ func (cw *curlWriter) string() string {
 }
 
 func (cw *curlWriter) addWithEsc(str string) {
-	escStr := `'` + strings.Replace(str, `'`, `'\''`, -1) + `'`
+	escStr := `'` + strings.ReplaceAll(str, `'`, `'\''`) + `'`
 	cw.add(escStr)
 }
 

app/vmalert/templates/template.go

@@ -198,8 +198,8 @@ func templateFuncs() textTpl.FuncMap {
 		// It is better to use quotesEscape, jsonEscape, queryEscape or pathEscape instead -
 		// these functions properly escape `\n` and `\r` chars according to their purpose.
 		"crlfEscape": func(q string) string {
-			q = strings.Replace(q, "\n", `\n`, -1)
-			return strings.Replace(q, "\r", `\r`, -1)
+			q = strings.ReplaceAll(q, "\n", `\n`)
+			return strings.ReplaceAll(q, "\r", `\r`)
 		},
 
 		// quotesEscape escapes the string, so it can be safely put inside JSON string.

app/vmalert/web.go

@@ -459,7 +459,7 @@ func (rh *requestHandler) listNotifiers() ([]byte, error) {
 		}
 		for _, target := range protoTargets {
 			notifier.Targets = append(notifier.Targets, &apiTarget{
-				Address: target.Notifier.Addr(),
+				Address: target.Addr(),
 				Labels:  target.Labels.ToMap(),
 			})
 		}

app/vmauth/auth_config.go

@@ -723,14 +723,11 @@ func reloadAuthConfigData(data []byte) (bool, error) {
 }
 
 func parseAuthConfig(data []byte) (*AuthConfig, error) {
-	data, err := envtemplate.ReplaceBytes(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot expand environment vars: %w", err)
-	}
+	data = envtemplate.ReplaceBytes(data)
 	ac := &AuthConfig{
 		ms: metrics.NewSet(),
 	}
-	if err = yaml.UnmarshalStrict(data, ac); err != nil {
+	if err := yaml.UnmarshalStrict(data, ac); err != nil {
 		return nil, fmt.Errorf("cannot unmarshal AuthConfig data: %w", err)
 	}
 

app/vmauth/main.go

@@ -110,9 +110,7 @@ func main() {
 	})
 
 	if len(*httpInternalListenAddr) > 0 {
-		go httpserver.Serve(*httpInternalListenAddr, internalRequestHandler, httpserver.ServeOptions{
-			UseProxyProtocol: useProxyProtocol,
-		})
+		go httpserver.Serve(*httpInternalListenAddr, internalRequestHandler, httpserver.ServeOptions{})
 	}
 	logger.Infof("started vmauth in %.3f seconds", time.Since(startTime).Seconds())
 

app/vmbackup/Makefile

@@ -1,106 +1,110 @@
 # All these commands must run from repository root.
 
+# special tag to reduce resulting binary size
+# See this issue https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8008
+VMBACKUP_GO_BUILD_TAGS=disable_grpc_modules
+
 vmbackup:
-	APP_NAME=vmbackup $(MAKE) app-local
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-local
 
 vmbackup-race:
-	APP_NAME=vmbackup RACE=-race $(MAKE) app-local
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) RACE=-race $(MAKE) app-local
 
 vmbackup-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker
 
 vmbackup-pure-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-pure
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-pure
 
 vmbackup-linux-amd64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-linux-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-linux-amd64
 
 vmbackup-linux-arm-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-linux-arm
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-linux-arm
 
 vmbackup-linux-arm64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-linux-arm64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-linux-arm64
 
 vmbackup-linux-ppc64le-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-linux-ppc64le
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-linux-ppc64le
 
 vmbackup-linux-386-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-linux-386
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-linux-386
 
 vmbackup-darwin-amd64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-darwin-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-darwin-amd64
 
 vmbackup-darwin-arm64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-darwin-arm64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-darwin-arm64
 
 vmbackup-freebsd-amd64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-freebsd-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-freebsd-amd64
 
 vmbackup-openbsd-amd64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-openbsd-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-openbsd-amd64
 
 vmbackup-windows-amd64-prod:
-	APP_NAME=vmbackup $(MAKE) app-via-docker-windows-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-via-docker-windows-amd64
 
 package-vmbackup:
-	APP_NAME=vmbackup $(MAKE) package-via-docker
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker
 
 package-vmbackup-pure:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-pure
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-pure
 
 package-vmbackup-amd64:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-amd64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-amd64
 
 package-vmbackup-arm:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-arm
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-arm
 
 package-vmbackup-arm64:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-arm64
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-arm64
 
 package-vmbackup-ppc64le:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-ppc64le
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-ppc64le
 
 package-vmbackup-386:
-	APP_NAME=vmbackup $(MAKE) package-via-docker-386
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) package-via-docker-386
 
 publish-vmbackup:
-	APP_NAME=vmbackup $(MAKE) publish-via-docker
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) publish-via-docker
 
 vmbackup-linux-amd64:
-	APP_NAME=vmbackup CGO_ENABLED=1 GOOS=linux GOARCH=amd64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=1 GOOS=linux GOARCH=amd64 $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-arm:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=arm $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=arm $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-arm64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=arm64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=arm64 $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-ppc64le:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=ppc64le $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=ppc64le $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-s390x:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=s390x $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=s390x $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-loong64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=loong64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=loong64 $(MAKE) app-local-goos-goarch
 
 vmbackup-linux-386:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=linux GOARCH=386 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=linux GOARCH=386 $(MAKE) app-local-goos-goarch
 
 vmbackup-darwin-amd64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 $(MAKE) app-local-goos-goarch
 
 vmbackup-darwin-arm64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 $(MAKE) app-local-goos-goarch
 
 vmbackup-freebsd-amd64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=freebsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
 
 vmbackup-openbsd-amd64:
-	APP_NAME=vmbackup CGO_ENABLED=0 GOOS=openbsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) CGO_ENABLED=0 GOOS=openbsd GOARCH=amd64 $(MAKE) app-local-goos-goarch
 
 vmbackup-windows-amd64:
-	GOARCH=amd64 APP_NAME=vmbackup $(MAKE) app-local-windows-goarch
+	GOARCH=amd64 APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-local-windows-goarch
 
 vmbackup-pure:
-	APP_NAME=vmbackup $(MAKE) app-local-pure
+	APP_NAME=vmbackup EXTRA_GO_BUILD_TAGS=$(VMBACKUP_GO_BUILD_TAGS) $(MAKE) app-local-pure

app/vmctl/flags.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/urfave/cli/v2"
@@ -383,6 +384,7 @@ const (
 	promFilterTimeEnd    = "prom-filter-time-end"
 	promFilterLabel      = "prom-filter-label"
 	promFilterLabelValue = "prom-filter-label-value"
+	promTemporaryDirPath = "prom-tmp-dir-path"
 )
 
 var (
@@ -414,6 +416,11 @@ var (
 			Usage: fmt.Sprintf("Prometheus regular expression to filter label from %q flag.", promFilterLabel),
 			Value: ".*",
 		},
+		&cli.StringFlag{
+			Name:  promTemporaryDirPath,
+			Usage: "Path to directory to be used for temporary files.",
+			Value: os.TempDir(),
+		},
 	}
 )
 

app/vmctl/main.go

@@ -252,7 +252,8 @@ func main() {
 					}
 
 					promCfg := prometheus.Config{
-						Snapshot: c.String(promSnapshot),
+						Snapshot:     c.String(promSnapshot),
+						TemporaryDir: c.String(promTemporaryDirPath),
 						Filter: prometheus.Filter{
 							TimeMin:    c.String(promFilterTimeStart),
 							TimeMax:    c.String(promFilterTimeEnd),

app/vmctl/opentsdb.go

@@ -132,7 +132,7 @@ func (op *otsdbProcessor) run() error {
 		close(errCh)
 		// check for any lingering errors on the query side
 		for otsdbErr := range errCh {
-			return fmt.Errorf("Import process failed: \n%s", otsdbErr)
+			return fmt.Errorf("import process failed: \n%s", otsdbErr)
 		}
 		bar.Finish()
 		log.Print(op.im.Stats())

app/vmctl/opentsdb/opentsdb.go

@@ -109,7 +109,7 @@ func (c Client) FindMetrics(q string) ([]string, error) {
 		return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
 	}
 	if resp.StatusCode != 200 {
-		return nil, fmt.Errorf("Bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
+		return nil, fmt.Errorf("bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	body, err := io.ReadAll(resp.Body)
@@ -133,7 +133,7 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
 		return nil, fmt.Errorf("failed to set GET request to %q: %s", q, err)
 	}
 	if resp.StatusCode != 200 {
-		return nil, fmt.Errorf("Bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
+		return nil, fmt.Errorf("bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	body, err := io.ReadAll(resp.Body)
@@ -310,7 +310,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 :: %v", r, err)
+			return &Client{}, fmt.Errorf("couldn't parse retention %q :: %v", r, err)
 		}
 		retentions = append(retentions, ret)
 	}

app/vmctl/prometheus/prometheus.go

@@ -16,7 +16,8 @@ type Config struct {
 	// Path to snapshot directory
 	Snapshot string
 
-	Filter Filter
+	Filter       Filter
+	TemporaryDir string
 }
 
 // Filter contains configuration for filtering
@@ -54,7 +55,7 @@ func (f filter) inRange(minV, maxV int64) bool {
 // NewClient creates and validates new Client
 // with given Config
 func NewClient(cfg Config) (*Client, error) {
-	db, err := tsdb.OpenDBReadOnly(cfg.Snapshot, "", nil)
+	db, err := tsdb.OpenDBReadOnly(cfg.Snapshot, cfg.TemporaryDir, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open snapshot %q: %s", cfg.Snapshot, err)
 	}

app/vmctl/stepper/split.go

@@ -33,7 +33,7 @@ func SplitDateRange(start, end time.Time, step string, timeReverse bool) ([][]ti
 	case StepMonth:
 		nextStep = func(t time.Time) (time.Time, time.Time) {
 			endOfMonth := time.Date(t.Year(), t.Month()+1, 1, 0, 0, 0, 0, t.Location()).Add(-1 * time.Nanosecond)
-			if t == endOfMonth {
+			if t.Equal(endOfMonth) {
 				endOfMonth = time.Date(t.Year(), t.Month()+2, 1, 0, 0, 0, 0, t.Location()).Add(-1 * time.Nanosecond)
 				t = time.Date(t.Year(), t.Month()+1, 1, 0, 0, 0, 0, t.Location())
 			}

app/vmctl/terminal/unix_bsd.go

@@ -1,4 +1,4 @@
-//go:build darwin || freebsd || openbsd
+//go:build darwin || freebsd || netbsd || openbsd
 
 package terminal
 

app/vmctl/vm_native.go

@@ -121,7 +121,7 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
 		pr := bar.NewProxyReader(reader)
 		if pr != nil {
 			reader = pr
-			fmt.Printf("Continue import process with filter %s:\n", f.String())
+			fmt.Fprintf(log.Writer(), "Continue import process with filter %s:\n", f.String())
 		}
 	}
 
@@ -191,7 +191,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 		initParams = []any{srcURL, dstURL, p.filter.String(), tenantID}
 	}
 
-	fmt.Println("") // extra line for better output formatting
+	fmt.Fprintln(log.Writer(), "") // extra line for better output formatting
 	log.Printf(initMessage, initParams...)
 	if len(ranges) > 1 {
 		log.Printf("Selected time range will be split into %d ranges according to %q step", len(ranges), p.filter.Chunk)

app/vminsert/common/insert_ctx.go

@@ -10,7 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -63,7 +63,7 @@ type InsertCtx struct {
 func (ctx *InsertCtx) Reset(rowsLen int) {
 	labels := ctx.Labels
 	for i := range labels {
-		labels[i] = prompbmarshal.Label{}
+		labels[i] = prompb.Label{}
 	}
 	ctx.Labels = labels[:0]
 
@@ -84,7 +84,7 @@ func cleanMetricRow(mr *storage.MetricRow) {
 	mr.MetricNameRaw = nil
 }
 
-func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompbmarshal.Label) []byte {
+func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompb.Label) []byte {
 	start := len(ctx.metricNamesBuf)
 	ctx.metricNamesBuf = append(ctx.metricNamesBuf, prefix...)
 	ctx.metricNamesBuf = storage.MarshalMetricNameRaw(ctx.metricNamesBuf, labels)
@@ -113,7 +113,7 @@ func (ctx *InsertCtx) TryPrepareLabels(hasRelabeling bool) bool {
 // WriteDataPoint writes (timestamp, value) with the given prefix and labels into ctx buffer.
 //
 // caller should invoke TryPrepareLabels before using this function if needed
-func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompbmarshal.Label, timestamp int64, value float64) error {
+func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompb.Label, timestamp int64, value float64) error {
 	metricNameRaw := ctx.marshalMetricNameRaw(prefix, labels)
 	return ctx.addRow(metricNameRaw, timestamp, value)
 }
@@ -123,7 +123,7 @@ func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompbmarshal.Label
 // caller must invoke TryPrepareLabels before using this function
 //
 // It returns metricNameRaw for the given labels if len(metricNameRaw) == 0.
-func (ctx *InsertCtx) WriteDataPointExt(metricNameRaw []byte, labels []prompbmarshal.Label, timestamp int64, value float64) ([]byte, error) {
+func (ctx *InsertCtx) WriteDataPointExt(metricNameRaw []byte, labels []prompb.Label, timestamp int64, value float64) ([]byte, error) {
 	if len(metricNameRaw) == 0 {
 		metricNameRaw = ctx.marshalMetricNameRaw(nil, labels)
 	}
@@ -161,7 +161,7 @@ func (ctx *InsertCtx) AddLabelBytes(name, value []byte) {
 		// Do not skip labels with empty name, since they are equal to __name__.
 		return
 	}
-	ctx.Labels = append(ctx.Labels, prompbmarshal.Label{
+	ctx.Labels = append(ctx.Labels, prompb.Label{
 		// Do not copy name and value contents for performance reasons.
 		// This reduces GC overhead on the number of objects and allocations.
 		Name:  bytesutil.ToUnsafeString(name),
@@ -179,7 +179,7 @@ func (ctx *InsertCtx) AddLabel(name, value string) {
 		// Do not skip labels with empty name, since they are equal to __name__.
 		return
 	}
-	ctx.Labels = append(ctx.Labels, prompbmarshal.Label{
+	ctx.Labels = append(ctx.Labels, prompb.Label{
 		// Do not copy name and value contents for performance reasons.
 		// This reduces GC overhead on the number of objects and allocations.
 		Name:  name,

app/vminsert/common/sort_labels.go

@@ -4,7 +4,7 @@ import (
 	"flag"
 	"sort"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 var sortLabels = flag.Bool("sortLabels", false, `Whether to sort labels for incoming samples before writing them to storage. `+
@@ -19,7 +19,7 @@ func (ctx *InsertCtx) SortLabelsIfNeeded() {
 	}
 }
 
-type sortedLabels []prompbmarshal.Label
+type sortedLabels []prompb.Label
 
 func (sl *sortedLabels) Len() int { return len(*sl) }
 func (sl *sortedLabels) Less(i, j int) bool {

app/vminsert/common/streamaggr.go

@@ -12,7 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 	"github.com/VictoriaMetrics/metrics"
@@ -59,7 +59,7 @@ func CheckStreamAggrConfig() error {
 	if *streamAggrConfig == "" {
 		return nil
 	}
-	pushNoop := func(_ []prompbmarshal.TimeSeries) {}
+	pushNoop := func(_ []prompb.TimeSeries) {}
 	opts := &streamaggr.Options{
 		DedupInterval:        *streamAggrDedupInterval,
 		DropInputLabels:      *streamAggrDropInputLabels,
@@ -170,9 +170,9 @@ func MustStopStreamAggr() {
 
 type streamAggrCtx struct {
 	mn      storage.MetricName
-	tss     []prompbmarshal.TimeSeries
-	labels  []prompbmarshal.Label
-	samples []prompbmarshal.Sample
+	tss     []prompb.TimeSeries
+	labels  []prompb.Label
+	samples []prompb.Sample
 	buf     []byte
 }
 
@@ -207,7 +207,7 @@ func (ctx *streamAggrCtx) push(mrs []storage.MetricRow, matchIdxs []byte) []byte
 		bufLen := len(buf)
 		buf = append(buf, mn.MetricGroup...)
 		metricGroup := bytesutil.ToUnsafeString(buf[bufLen:])
-		labels = append(labels, prompbmarshal.Label{
+		labels = append(labels, prompb.Label{
 			Name:  "__name__",
 			Value: metricGroup,
 		})
@@ -220,19 +220,19 @@ func (ctx *streamAggrCtx) push(mrs []storage.MetricRow, matchIdxs []byte) []byte
 			bufLen = len(buf)
 			buf = append(buf, tag.Value...)
 			value := bytesutil.ToUnsafeString(buf[bufLen:])
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  name,
 				Value: value,
 			})
 		}
 
 		samplesLen := len(samples)
-		samples = append(samples, prompbmarshal.Sample{
+		samples = append(samples, prompb.Sample{
 			Timestamp: mr.Timestamp,
 			Value:     mr.Value,
 		})
 
-		tss = append(tss, prompbmarshal.TimeSeries{
+		tss = append(tss, prompb.TimeSeries{
 			Labels:  labels[labelsLen:],
 			Samples: samples[samplesLen:],
 		})
@@ -260,7 +260,7 @@ func (ctx *streamAggrCtx) push(mrs []storage.MetricRow, matchIdxs []byte) []byte
 	return matchIdxs
 }
 
-func pushAggregateSeries(tss []prompbmarshal.TimeSeries) {
+func pushAggregateSeries(tss []prompb.TimeSeries) {
 	currentTimestamp := int64(fasttime.UnixTimestamp()) * 1000
 	var ctx InsertCtx
 	ctx.Reset(len(tss))

app/vminsert/csvimport/request_handler.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -28,7 +28,7 @@ func InsertHandler(req *http.Request) error {
 	})
 }
 
-func insertRows(rows []csvimport.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(rows []csvimport.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/datadogsketches/request_handler.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
@@ -30,7 +30,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	})
 }
 
-func insertRows(sketches []*datadogsketches.Sketch, extraLabels []prompbmarshal.Label) error {
+func insertRows(sketches []*datadogsketches.Sketch, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/datadogv1/request_handler.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1/stream"
@@ -30,7 +30,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	})
 }
 
-func insertRows(series []datadogv1.Series, extraLabels []prompbmarshal.Label) error {
+func insertRows(series []datadogv1.Series, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/datadogv2/request_handler.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2/stream"
@@ -33,7 +33,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	})
 }
 
-func insertRows(series []datadogv2.Series, extraLabels []prompbmarshal.Label) error {
+func insertRows(series []datadogv2.Series, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/influx/request_handler.go

@@ -9,7 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -58,7 +58,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	})
 }
 
-func insertRows(db string, rows []influx.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(db string, rows []influx.Row, extraLabels []prompb.Label) error {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)
 
@@ -157,7 +157,7 @@ type pushCtx struct {
 	Common         common.InsertCtx
 	metricNameBuf  []byte
 	metricGroupBuf []byte
-	originLabels   []prompbmarshal.Label
+	originLabels   []prompb.Label
 }
 
 func (ctx *pushCtx) reset() {
@@ -167,7 +167,7 @@ func (ctx *pushCtx) reset() {
 
 	originLabels := ctx.originLabels
 	for i := range originLabels {
-		originLabels[i] = prompbmarshal.Label{}
+		originLabels[i] = prompb.Label{}
 	}
 	ctx.originLabels = originLabels[:0]
 }

app/vminsert/main.go

@@ -37,7 +37,7 @@ import (
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
 	opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -101,7 +101,7 @@ func Init() {
 	if len(*opentsdbHTTPListenAddr) > 0 {
 		opentsdbhttpServer = opentsdbhttpserver.MustStart(*opentsdbHTTPListenAddr, *opentsdbHTTPUseProxyProtocol, opentsdbhttp.InsertHandler)
 	}
-	promscrape.Init(func(_ *auth.Token, wr *prompbmarshal.WriteRequest) {
+	promscrape.Init(func(_ *auth.Token, wr *prompb.WriteRequest) {
 		prompush.Push(wr)
 	})
 	timeserieslimits.Init(*maxLabelsPerTimeseries, *maxLabelNameLen, *maxLabelValueLen)
@@ -131,7 +131,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	startTime := time.Now()
 	defer requestDuration.UpdateDuration(startTime)
 
-	path := strings.Replace(r.URL.Path, "//", "/", -1)
+	path := strings.ReplaceAll(r.URL.Path, "//", "/")
 	if strings.HasPrefix(path, "/static") {
 		staticServer.ServeHTTP(w, r)
 		return true

app/vminsert/native/request_handler.go

@@ -7,7 +7,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -31,7 +31,7 @@ func InsertHandler(req *http.Request) error {
 	})
 }
 
-func insertRows(block *stream.Block, extraLabels []prompbmarshal.Label) error {
+func insertRows(block *stream.Block, extraLabels []prompb.Label) error {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)
 

app/vminsert/newrelic/request_handler.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -30,7 +30,7 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	})
 }
 
-func insertRows(rows []newrelic.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(rows []newrelic.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/opentelemetry/request_handler.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -33,12 +33,12 @@ func InsertHandler(req *http.Request) error {
 			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
 		}
 	}
-	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompbmarshal.TimeSeries) error {
+	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompb.TimeSeries, _ []prompb.MetricMetadata) error {
 		return insertRows(tss, extraLabels)
 	})
 }
 
-func insertRows(tss []prompbmarshal.TimeSeries, extraLabels []prompbmarshal.Label) error {
+func insertRows(tss []prompb.TimeSeries, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/opentsdbhttp/request_handler.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -36,7 +36,7 @@ func InsertHandler(req *http.Request) error {
 	}
 }
 
-func insertRows(rows []opentsdbhttp.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(rows []opentsdbhttp.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/prometheusimport/request_handler.go

@@ -6,7 +6,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -29,14 +30,14 @@ func InsertHandler(req *http.Request) error {
 		return err
 	}
 	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, defaultTimestamp, encoding, true, func(rows []prometheus.Row) error {
+	return stream.Parse(req.Body, defaultTimestamp, encoding, true, promscrape.IsMetadataEnabled(), func(rows []prometheus.Row, _ []prometheus.Metadata) error {
 		return insertRows(rows, extraLabels)
 	}, func(s string) {
 		httpserver.LogError(req, s)
 	})
 }
 
-func insertRows(rows []prometheus.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(rows []prometheus.Row, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/prompush/push.go

@@ -4,7 +4,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -16,7 +16,7 @@ var (
 const maxRowsPerBlock = 10000
 
 // Push pushes wr for the given at to storage.
-func Push(wr *prompbmarshal.WriteRequest) {
+func Push(wr *prompb.WriteRequest) {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 
@@ -43,7 +43,7 @@ func Push(wr *prompbmarshal.WriteRequest) {
 	}
 }
 
-func push(ctx *common.InsertCtx, tss []prompbmarshal.TimeSeries) {
+func push(ctx *common.InsertCtx, tss []prompb.TimeSeries) {
 	rowsLen := 0
 	for i := range tss {
 		rowsLen += len(tss[i].Samples)

app/vminsert/promremotewrite/request_handler.go

@@ -6,7 +6,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
@@ -24,12 +23,12 @@ func InsertHandler(req *http.Request) error {
 		return err
 	}
 	isVMRemoteWrite := req.Header.Get("Content-Encoding") == "zstd"
-	return stream.Parse(req.Body, isVMRemoteWrite, func(tss []prompb.TimeSeries) error {
+	return stream.Parse(req.Body, isVMRemoteWrite, func(tss []prompb.TimeSeries, _ []prompb.MetricMetadata) error {
 		return insertRows(tss, extraLabels)
 	})
 }
 
-func insertRows(timeseries []prompb.TimeSeries, extraLabels []prompbmarshal.Label) error {
+func insertRows(timeseries []prompb.TimeSeries, extraLabels []prompb.Label) error {
 	ctx := common.GetInsertCtx()
 	defer common.PutInsertCtx(ctx)
 

app/vminsert/relabel/relabel.go

@@ -8,7 +8,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -102,7 +102,7 @@ func HasRelabeling() bool {
 // Ctx holds relabeling context.
 type Ctx struct {
 	// tmpLabels is used during ApplyRelabeling call.
-	tmpLabels []prompbmarshal.Label
+	tmpLabels []prompb.Label
 }
 
 // Reset resets ctx.
@@ -114,13 +114,13 @@ func (ctx *Ctx) Reset() {
 // ApplyRelabeling applies relabeling to the given labels and returns the result.
 //
 // The returned labels are valid until the next call to ApplyRelabeling.
-func (ctx *Ctx) ApplyRelabeling(labels []prompbmarshal.Label) []prompbmarshal.Label {
+func (ctx *Ctx) ApplyRelabeling(labels []prompb.Label) []prompb.Label {
 	pcs := pcsGlobal.Load()
 	if pcs.Len() == 0 && !*usePromCompatibleNaming {
 		// There are no relabeling rules.
 		return labels
 	}
-	// Convert labels to prompbmarshal.Label format suitable for relabeling.
+	// Convert labels to prompb.Label format suitable for relabeling.
 	tmpLabels := ctx.tmpLabels[:0]
 	for _, label := range labels {
 		name := label.Name
@@ -128,7 +128,7 @@ func (ctx *Ctx) ApplyRelabeling(labels []prompbmarshal.Label) []prompbmarshal.La
 			name = "__name__"
 		}
 		value := label.Value
-		tmpLabels = append(tmpLabels, prompbmarshal.Label{
+		tmpLabels = append(tmpLabels, prompb.Label{
 			Name:  name,
 			Value: value,
 		})
@@ -165,7 +165,7 @@ func (ctx *Ctx) ApplyRelabeling(labels []prompbmarshal.Label) []prompbmarshal.La
 			name = ""
 		}
 		value := label.Value
-		dst = append(dst, prompbmarshal.Label{
+		dst = append(dst, prompb.Label{
 			Name:  name,
 			Value: value,
 		})

app/vminsert/vmimport/request_handler.go

@@ -7,7 +7,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport/stream"
@@ -34,7 +34,7 @@ func InsertHandler(req *http.Request) error {
 	})
 }
 
-func insertRows(rows []vmimport.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(rows []vmimport.Row, extraLabels []prompb.Label) error {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)