add changelog

.github/workflows/main.yml

@@ -90,7 +90,7 @@ jobs:
       - name: Publish coverage
         uses: codecov/codecov-action@v5
         with:
-          files: ./coverage.txt
+          file: ./coverage.txt
 
   integration-test:
     name: integration-test

.github/workflows/sync-docs.yml

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

Makefile

@@ -18,7 +18,7 @@ TAR_OWNERSHIP ?= --owner=1000 --group=1000
 .PHONY: $(MAKECMDGOALS)
 
 include app/*/Makefile
-include codespell/Makefile
+include cspell/Makefile
 include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
@@ -567,7 +567,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run
 
 install-golangci-lint:
-	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.7
+	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.63.4
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`

README.md

@@ -3,7 +3,7 @@
 ![Latest Release](https://img.shields.io/github/v/release/VictoriaMetrics/VictoriaMetrics?sort=semver&label=&filter=!*-victorialogs&logo=github&labelColor=gray&color=gray&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Freleases%2Flatest)
 ![Docker Pulls](https://img.shields.io/docker/pulls/victoriametrics/victoria-metrics?label=&logo=docker&logoColor=white&labelColor=2496ED&color=2496ED&link=https%3A%2F%2Fhub.docker.com%2Fr%2Fvictoriametrics%2Fvictoria-metrics)
 ![Go Report](https://goreportcard.com/badge/github.com/VictoriaMetrics/VictoriaMetrics?link=https%3A%2F%2Fgoreportcard.com%2Freport%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics)
-![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/main.yml/badge.svg?branch=master&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Factions)
+![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/workflows/main/badge.svg?link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Factions)
 ![codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/branch/master/graph/badge.svg?link=https%3A%2F%2Fcodecov.io%2Fgh%2FVictoriaMetrics%2FVictoriaMetrics)
 ![License](https://img.shields.io/github/license/VictoriaMetrics/VictoriaMetrics?labelColor=green&label=&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Fblob%2Fmaster%2FLICENSE)
 ![Slack](https://img.shields.io/badge/Join-4A154B?logo=slack&link=https%3A%2F%2Fslack.victoriametrics.com)
@@ -22,7 +22,7 @@ Here are some resources and information about VictoriaMetrics:
 
 - Documentation: [docs.victoriametrics.com](https://docs.victoriametrics.com)
 - Case studies: [Grammarly, Roblox, Wix,...](https://docs.victoriametrics.com/casestudies/).
-- Available: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), docker images [Docker Hub](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [Quay](https://quay.io/repository/victoriametrics/victoria-metrics), [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics)
+- Available: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), [Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/), [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics)
 - Deployment types: [Single-node version](https://docs.victoriametrics.com/), [Cluster version](https://docs.victoriametrics.com/cluster-victoriametrics/), and [Enterprise version](https://docs.victoriametrics.com/enterprise/)
 - Changelog: [CHANGELOG](https://docs.victoriametrics.com/changelog/), and [How to upgrade](https://docs.victoriametrics.com/#how-to-upgrade-victoriametrics)
 - Community: [Slack](https://slack.victoriametrics.com/), [X (Twitter)](https://x.com/VictoriaMetrics), [LinkedIn](https://www.linkedin.com/company/victoriametrics/), [YouTube](https://www.youtube.com/@VictoriaMetrics)

SECURITY.md

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

app/vlinsert/datadog/datadog.go

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

app/vlinsert/elasticsearch/elasticsearch.go

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

app/vlinsert/elasticsearch/elasticsearch_test.go

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

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

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

app/vlinsert/insertutils/common_params.go

@@ -141,7 +141,7 @@ type LogMessageProcessor interface {
 	//
 	// If streamFields is non-nil, then the given streamFields must be used as log stream fields instead of pre-configured fields.
 	//
-	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can reuse them.
+	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can re-use them.
 	AddRow(timestamp int64, fields, streamFields []logstorage.Field)
 
 	// MustClose() must flush all the remaining fields and free up resources occupied by LogMessageProcessor.

app/vlinsert/insertutils/timestamp.go

@@ -2,9 +2,7 @@ package insertutils
 
 import (
 	"fmt"
-	"math"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -51,35 +49,6 @@ func parseTimestamp(s string) (int64, error) {
 
 // ParseUnixTimestamp parses s as unix timestamp in seconds, milliseconds, microseconds or nanoseconds and returns the parsed timestamp in nanoseconds.
 func ParseUnixTimestamp(s string) (int64, error) {
-	if strings.IndexByte(s, '.') >= 0 {
-		// Parse timestamp as floating-point value
-		f, err := strconv.ParseFloat(s, 64)
-		if err != nil {
-			return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)
-		}
-		if f < (1<<31) && f >= (-1<<31) {
-			// The timestamp is in seconds.
-			return int64(f * 1e9), nil
-		}
-		if f < 1e3*(1<<31) && f >= 1e3*(-1<<31) {
-			// The timestamp is in milliseconds.
-			return int64(f * 1e6), nil
-		}
-		if f < 1e6*(1<<31) && f >= 1e6*(-1<<31) {
-			// The timestamp is in microseconds.
-			return int64(f * 1e3), nil
-		}
-		// The timestamp is in nanoseconds
-		if f > math.MaxInt64 {
-			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, int64(math.MaxInt64))
-		}
-		if f < math.MinInt64 {
-			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, int64(math.MinInt64))
-		}
-		return int64(f), nil
-	}
-
-	// Parse timestamp as integer
 	n, err := strconv.ParseInt(s, 10, 64)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)

app/vlinsert/insertutils/timestamp_test.go

@@ -6,63 +6,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
-func TestParseUnixTimestamp_Success(t *testing.T) {
-	f := func(s string, timestampExpected int64) {
-		t.Helper()
-
-		timestamp, err := ParseUnixTimestamp(s)
-		if err != nil {
-			t.Fatalf("unexpected error in ParseUnixTimestamp(%q): %s", s, err)
-		}
-		if timestamp != timestampExpected {
-			t.Fatalf("unexpected timestamp returned from ParseUnixTimestamp(%q); got %d; want %d", s, timestamp, timestampExpected)
-		}
-	}
-
-	f("0", 0)
-
-	// nanoseconds
-	f("-1234567890123456789", -1234567890123456789)
-	f("1234567890123456789", 1234567890123456789)
-
-	// microseconds
-	f("-1234567890123456", -1234567890123456000)
-	f("1234567890123456", 1234567890123456000)
-	f("1234567890123456.789", 1234567890123456768)
-
-	// milliseconds
-	f("-1234567890123", -1234567890123000000)
-	f("1234567890123", 1234567890123000000)
-	f("1234567890123.456", 1234567890123456000)
-
-	// seconds
-	f("-1234567890", -1234567890000000000)
-	f("1234567890", 1234567890000000000)
-	f("-1234567890.123456", -1234567890123456000)
-}
-
-func TestParseUnixTimestamp_Failure(t *testing.T) {
-	f := func(s string) {
-		t.Helper()
-
-		_, err := ParseUnixTimestamp(s)
-		if err == nil {
-			t.Fatalf("expecting non-nil error in ParseUnixTimestamp(%q)", s)
-		}
-	}
-
-	// non-numeric timestamp
-	f("")
-	f("foobar")
-	f("foo.bar")
-
-	// too big timestamp
-	f("12345678901234567890")
-	f("-12345678901234567890")
-	f("12345678901234567890.235424")
-	f("-12345678901234567890.235424")
-}
-
 func TestExtractTimestampFromFields_Success(t *testing.T) {
 	f := func(timeField string, fields []logstorage.Field, nsecsExpected int64) {
 		t.Helper()

app/vlinsert/journald/journald.go

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

app/vlinsert/journald/journald_test.go

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

app/vlinsert/jsonline/jsonline.go

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

app/vlinsert/jsonline/jsonline_test.go

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

app/vlinsert/loki/loki.go

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

app/vlinsert/loki/loki_json.go

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

app/vlinsert/loki/loki_json_test.go

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

app/vlinsert/loki/loki_json_timing_test.go

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

app/vlinsert/loki/loki_protobuf.go

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

app/vlinsert/loki/loki_protobuf_test.go

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

app/vlinsert/loki/loki_protobuf_timing_test.go

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

app/vlinsert/main.go

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

app/vlinsert/opentelemetry/opentelemetry.go

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

app/vlinsert/opentelemetry/opentelemetry_test.go

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

app/vlinsert/syslog/syslog.go

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

app/vlogscli/json_prettifier.go

@@ -154,7 +154,7 @@ func readNextJSONObject(d *json.Decoder) ([]logstorage.Field, error) {
 		}
 		value, ok := t.(string)
 		if !ok {
-			return nil, fmt.Errorf("unexpected token read for object value: %v; want string", t)
+			return nil, fmt.Errorf("unexpected token read for oject value: %v; want string", t)
 		}
 
 		fields = append(fields, logstorage.Field{

app/vlogsgenerator/main.go

@@ -18,7 +18,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 var (
@@ -31,7 +31,7 @@ var (
 	totalStreams  = flag.Int("totalStreams", 0, "The number of total log streams; if -totalStreams > -activeStreams, then some active streams are substituted with new streams "+
 		"during data generation")
 	logsPerStream     = flag.Int64("logsPerStream", 1_000, "The number of log entries to generate per each log stream. Log entries are evenly distributed between -start and -end")
-	constFieldsPerLog = flag.Int("constFieldsPerLog", 3, "The number of fields with constant values to generate per each log entry; "+
+	constFieldsPerLog = flag.Int("constFieldsPerLog", 3, "The number of fields with constaint values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
 	varFieldsPerLog = flag.Int("varFieldsPerLog", 1, "The number of fields with variable values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
@@ -177,10 +177,7 @@ func generateAndPushLogs(cfg *workerConfig, workerID int) {
 	sw := &statWriter{
 		w: pw,
 	}
-
-	// The 1MB write buffer increases data ingestion performance by reducing the number of send() syscalls
-	bw := bufio.NewWriterSize(sw, 1024*1024)
-
+	bw := bufio.NewWriter(sw)
 	doneCh := make(chan struct{})
 	go func() {
 		generateLogs(bw, workerID, cfg.activeStreams, cfg.totalStreams)
@@ -309,7 +306,7 @@ type timeFlag struct {
 }
 
 func (tf *timeFlag) Set(s string) error {
-	msec, err := timeutil.ParseTimeMsec(s)
+	msec, err := promutils.ParseTimeMsec(s)
 	if err != nil {
 		return fmt.Errorf("cannot parse time from %q: %w", s, err)
 	}

app/vlselect/logsql/logsql.go

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

app/vlselect/main.go

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

app/vlselect/vmui/asset-manifest.json

@@ -0,0 +1,12 @@
+{
+  "files": {
+    "main.css": "./static/css/main.02a1c6cb.css",
+    "main.js": "./static/js/main.55c8060b.js",
+    "static/js/685.f772060c.chunk.js": "./static/js/685.f772060c.chunk.js",
+    "index.html": "./index.html"
+  },
+  "entrypoints": [
+    "static/css/main.02a1c6cb.css",
+    "static/js/main.55c8060b.js"
+  ]
+}

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

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

app/vlselect/vmui/index.html

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

app/vlselect/vmui/static/js/main.55c8060b.js.LICENSE.txt

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

app/vmagent/graphite/request_handler.go

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

app/vmagent/influx/request_handler.go

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

app/vmagent/main.go

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

app/vmagent/native/request_handler.go

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

app/vmagent/newrelic/request_handler.go

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

app/vmagent/opentelemetry/request_handler.go

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

app/vmagent/prometheusimport/request_handler.go

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

app/vmagent/prometheusimport/request_handler_test.go

@@ -26,7 +26,7 @@ func TestInsertHandler(t *testing.T) {
 	req := httptest.NewRequest(http.MethodPost, "/insert/0/api/v1/import/prometheus", bytes.NewBufferString(`{"foo":"bar"}
 go_memstats_alloc_bytes_total 1`))
 	if err := InsertHandler(nil, req); err != nil {
-		t.Fatalf("unexpected error %s", err)
+		t.Fatalf("unxepected error %s", err)
 	}
 	expectedMsg := "cannot unmarshal Prometheus line"
 	if !strings.Contains(testOutput.String(), expectedMsg) {

app/vmagent/remotewrite/pendingseries.go

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

app/vmagent/remotewrite/remotewrite_test.go

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

app/vmagent/remotewrite/streamaggr.go

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

app/vmagent/vmimport/request_handler.go

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

app/vmalert-tool/main.go

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

app/vmalert-tool/unittest/unittest.go

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

app/vmalert-tool/unittest/unittest_test.go

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

app/vmalert/config/config.go

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

app/vmalert/config/config_test.go

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

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

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

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

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

app/vmalert/config/types.go

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

app/vmalert/main.go

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

app/vmalert/notifier/alertmanager.go

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

app/vmalert/notifier/init.go

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

app/vmalert/notifier/notifier_blackhole.go

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

app/vmalert/rule/alerting.go

@@ -9,8 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -59,69 +57,6 @@ type alertingRuleMetrics struct {
 	seriesFetched *utils.Gauge
 }
 
-func newAlertingRuleMetrics(set *metrics.Set, ar *AlertingRule) *alertingRuleMetrics {
-	labels := fmt.Sprintf(`alertname=%q, group=%q, file=%q, id="%d"`, ar.Name, ar.GroupName, ar.File, ar.ID())
-	arm := &alertingRuleMetrics{}
-
-	arm.pending = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerts_pending{%s}`, labels),
-		func() float64 {
-			ar.alertsMu.RLock()
-			defer ar.alertsMu.RUnlock()
-			var num int
-			for _, a := range ar.alerts {
-				if a.State == notifier.StatePending {
-					num++
-				}
-			}
-			return float64(num)
-		})
-	arm.active = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
-		func() float64 {
-			ar.alertsMu.RLock()
-			defer ar.alertsMu.RUnlock()
-			var num int
-			for _, a := range ar.alerts {
-				if a.State == notifier.StateFiring {
-					num++
-				}
-			}
-			return float64(num)
-		})
-	arm.errors = utils.NewCounter(set, fmt.Sprintf(`vmalert_alerting_rules_errors_total{%s}`, labels))
-	arm.samples = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_samples{%s}`, labels),
-		func() float64 {
-			e := ar.state.getLast()
-			return float64(e.Samples)
-		})
-	arm.seriesFetched = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_series_fetched{%s}`, labels),
-		func() float64 {
-			e := ar.state.getLast()
-			if e.SeriesFetched == nil {
-				// means seriesFetched is unsupported
-				return -1
-			}
-			seriesFetched := float64(*e.SeriesFetched)
-			if seriesFetched == 0 && e.Samples > 0 {
-				// `alert: 0.95` will fetch no series
-				// but will get one time series in response.
-				seriesFetched = float64(e.Samples)
-			}
-			return seriesFetched
-		})
-	return arm
-}
-
-func (arm *alertingRuleMetrics) close() {
-	if arm == nil {
-		return
-	}
-	arm.errors.Unregister()
-	arm.active.Unregister()
-	arm.pending.Unregister()
-	arm.samples.Unregister()
-	arm.seriesFetched.Unregister()
-}
-
 // NewAlertingRule creates a new AlertingRule
 func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *AlertingRule {
 	ar := &AlertingRule{
@@ -146,7 +81,8 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 			Headers:                   group.Headers,
 			Debug:                     cfg.Debug,
 		}),
-		alerts: make(map[uint64]*notifier.Alert),
+		alerts:  make(map[uint64]*notifier.Alert),
+		metrics: &alertingRuleMetrics{},
 	}
 
 	entrySize := *ruleUpdateEntriesLimit
@@ -159,17 +95,63 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 	ar.state = &ruleState{
 		entries: make([]StateEntry, entrySize),
 	}
-	ar.metrics = newAlertingRuleMetrics(group.metrics.set, ar)
+
+	labels := fmt.Sprintf(`alertname=%q, group=%q, file=%q, id="%d"`, ar.Name, group.Name, group.File, ar.ID())
+	ar.metrics.pending = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerts_pending{%s}`, labels),
+		func() float64 {
+			ar.alertsMu.RLock()
+			defer ar.alertsMu.RUnlock()
+			var num int
+			for _, a := range ar.alerts {
+				if a.State == notifier.StatePending {
+					num++
+				}
+			}
+			return float64(num)
+		})
+	ar.metrics.active = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
+		func() float64 {
+			ar.alertsMu.RLock()
+			defer ar.alertsMu.RUnlock()
+			var num int
+			for _, a := range ar.alerts {
+				if a.State == notifier.StateFiring {
+					num++
+				}
+			}
+			return float64(num)
+		})
+	ar.metrics.errors = utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_alerting_rules_errors_total{%s}`, labels))
+	ar.metrics.samples = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_samples{%s}`, labels),
+		func() float64 {
+			e := ar.state.getLast()
+			return float64(e.Samples)
+		})
+	ar.metrics.seriesFetched = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_series_fetched{%s}`, labels),
+		func() float64 {
+			e := ar.state.getLast()
+			if e.SeriesFetched == nil {
+				// means seriesFetched is unsupported
+				return -1
+			}
+			seriesFetched := float64(*e.SeriesFetched)
+			if seriesFetched == 0 && e.Samples > 0 {
+				// `alert: 0.95` will fetch no series
+				// but will get one time series in response.
+				seriesFetched = float64(e.Samples)
+			}
+			return seriesFetched
+		})
 	return ar
 }
 
-func (ar *AlertingRule) registerMetrics(g *Group) {
-	ar.metrics = newAlertingRuleMetrics(g.metrics.set, ar)
-}
-
 // close unregisters rule metrics
-func (ar *AlertingRule) unregisterMetrics() {
-	ar.metrics.close()
+func (ar *AlertingRule) close() {
+	ar.metrics.active.Unregister()
+	ar.metrics.pending.Unregister()
+	ar.metrics.errors.Unregister()
+	ar.metrics.samples.Unregister()
+	ar.metrics.seriesFetched.Unregister()
 }
 
 // String implements Stringer interface

app/vmalert/rule/alerting_test.go

@@ -11,8 +11,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -930,7 +928,7 @@ func TestGroup_Restore(t *testing.T) {
 			},
 		})
 
-	// one active alert with restore labels mismatch
+	// one active alert with restore labels missmatch
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev", "team", "foo"))
@@ -975,7 +973,7 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 		t.Fatalf("expected to get err; got nil")
 	}
 	if !strings.Contains(err.Error(), expErr) {
-		t.Fatalf("expected to get err %q; got %q instead", expErr, err)
+		t.Fatalf("expected to get err %q; got %q insterad", expErr, err)
 	}
 }
 
@@ -1250,17 +1248,13 @@ func newTestAlertingRule(name string, waitFor time.Duration) *AlertingRule {
 		EvalInterval: waitFor,
 		alerts:       make(map[uint64]*notifier.Alert),
 		state:        &ruleState{entries: make([]StateEntry, 10)},
-		metrics:      getTestAlertingRuleMetrics(name),
+		metrics: &alertingRuleMetrics{
+			errors: utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name)),
+		},
 	}
 	return &rule
 }
 
-func getTestAlertingRuleMetrics(name string) *alertingRuleMetrics {
-	m := &alertingRuleMetrics{}
-	m.errors = utils.NewCounter(metrics.NewSet(), fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name))
-	return m
-}
-
 func newTestAlertingRuleWithCustomFields(name string, waitFor, evalInterval, keepFiringFor time.Duration, annotation map[string]string) *AlertingRule {
 	rule := newTestAlertingRule(name, waitFor)
 	if evalInterval != 0 {

app/vmalert/rule/group.go

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

app/vmalert/rule/group_test.go

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

app/vmalert/rule/recording.go

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

app/vmalert/rule/recording_test.go

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

app/vmalert/rule/rule.go

@@ -27,10 +27,9 @@ type Rule interface {
 	// updateWith performs modification of current Rule
 	// with fields of the given Rule.
 	updateWith(Rule) error
-	// unregister Rule metrics
-	unregisterMetrics()
-	// register Rule metrics with the given group
-	registerMetrics(g *Group)
+	// close performs the shutdown procedures for rule
+	// such as metrics unregister
+	close()
 }
 
 var errDuplicate = errors.New("result contains metrics with the same labelset during evaluation. See https://docs.victoriametrics.com/vmalert/#series-with-the-same-labelset for details")

app/vmalert/templates/template.go

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

app/vmalert/utils/metrics.go

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

app/vmalert/web.qtpl

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

app/vmalert/web.qtpl.go

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

app/vmalert/web_test.go

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

app/vmalert/web_types_test.go

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

app/vmauth/auth_config.go

@@ -141,7 +141,7 @@ func (h *Header) UnmarshalYAML(f func(any) error) error {
 
 	n := strings.IndexByte(s, ':')
 	if n < 0 {
-		return fmt.Errorf("missing separator char ':' between Name and Value in the header %q; expected format - 'Name: Value'", s)
+		return fmt.Errorf("missing speparator char ':' between Name and Value in the header %q; expected format - 'Name: Value'", s)
 	}
 	h.Name = strings.TrimSpace(s[:n])
 	h.Value = strings.TrimSpace(s[n+1:])
@@ -173,7 +173,7 @@ type URLMap struct {
 	// DiscoverBackendIPs instructs discovering URLPrefix backend IPs via DNS.
 	DiscoverBackendIPs *bool `yaml:"discover_backend_ips,omitempty"`
 
-	// HeadersConf is the config for augmenting request and response headers.
+	// HeadersConf is the config for augumenting request and response headers.
 	HeadersConf HeadersConf `yaml:",inline"`
 
 	// RetryStatusCodes is the list of response status codes used for retrying requests.
@@ -438,7 +438,7 @@ func getFirstAvailableBackendURL(bus []*backendURL) *backendURL {
 		return bu
 	}
 
-	// Slow path - the first url is temporarily unavailable. Fall back to the remaining urls.
+	// Slow path - the first url is temporarily unavailabel. Fall back to the remaining urls.
 	for i := 1; i < len(bus); i++ {
 		if !bus[i].isBroken() {
 			bu = bus[i]

app/vmauth/target_url_test.go

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

app/vmctl/flags.go

@@ -344,7 +344,7 @@ var (
 		},
 		&cli.BoolFlag{
 			Name:  influxSkipDatabaseLabel,
-			Usage: "Whether to skip adding the label 'db' to timeseries.",
+			Usage: "Wether to skip adding the label 'db' to timeseries.",
 			Value: false,
 		},
 		&cli.BoolFlag{

app/vmctl/influx/influx.go

@@ -144,9 +144,9 @@ func timeFilter(start, end string) string {
 // by checking available (non-empty) tags, fields and measurements
 // which unique combination represents all possible
 // time series existing in database.
-// Explore is required to reduce the load on influx
+// The explore required to reduce the load on influx
 // by querying field of the exact time series at once,
-// instead of fetching all the values over and over.
+// instead of fetching all of the values over and over.
 //
 // May contain non-existing time series.
 func (c *Client) Explore() ([]*Series, error) {
@@ -340,7 +340,10 @@ func (c *Client) fieldsByMeasurement() (map[string][]string, error) {
 }
 
 func (c *Client) getSeries() ([]*Series, error) {
-	com := c.getSeriesCommand()
+	com := "show series"
+	if c.filterSeries != "" {
+		com = fmt.Sprintf("%s %s", com, c.filterSeries)
+	}
 	q := influx.Query{
 		Command:         com,
 		Database:        c.database,
@@ -386,21 +389,6 @@ func (c *Client) getSeries() ([]*Series, error) {
 	return result, nil
 }
 
-func (c *Client) getSeriesCommand() string {
-	com := "show series"
-	if c.filterSeries != "" {
-		com = fmt.Sprintf("%s %s", com, c.filterSeries)
-	}
-	if c.filterTime != "" {
-		joinStatement := " where "
-		if strings.Contains(strings.ToLower(com), joinStatement) {
-			joinStatement = " AND "
-		}
-		com = fmt.Sprintf("%s%s%s", com, joinStatement, c.filterTime)
-	}
-	return com
-}
-
 // getMeasurementTags get the tags for each measurement.
 // tags are placed in a map without values (similar to a set) for quick lookups:
 // {"measurement1": {"tag1", "tag2"}, "measurement2": {"tag3", "tag4"}}

app/vmctl/influx/influx_test.go

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

app/vmctl/main.go

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

app/vmctl/testdata/servers_integration_test/remote_read_server.go

@@ -442,7 +442,7 @@ func (si *mockSamplesIterator) At() (int64, float64) {
 }
 
 func (si *mockSamplesIterator) AtHistogram(*histogram.Histogram) (int64, *histogram.Histogram) {
-	panic("BUG: mustn't be called")
+	panic("BUG: musn't be called")
 }
 
 func (si *mockSamplesIterator) AtFloatHistogram(*histogram.FloatHistogram) (int64, *histogram.FloatHistogram) {

app/vmctl/testdata/servers_integration_test/remote_write_server.go

@@ -212,7 +212,7 @@ func (rws *RemoteWriteServer) importNativeHandler(t *testing.T) http.Handler {
 		var gotTimeSeries []vm.TimeSeries
 		var mx sync.RWMutex
 
-		err := stream.Parse(r.Body, "", func(block *stream.Block) error {
+		err := stream.Parse(r.Body, false, func(block *stream.Block) error {
 			mn := &block.MetricName
 			var timeseries vm.TimeSeries
 			timeseries.Name = string(mn.MetricGroup)

app/vmctl/utils/time.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 const (
@@ -16,7 +16,7 @@ const (
 // ParseTime parses time in s string and returns time.Time object
 // if parse correctly or error if not
 func ParseTime(s string) (time.Time, error) {
-	msecs, err := timeutil.ParseTimeMsec(s)
+	msecs, err := promutils.ParseTimeMsec(s)
 	if err != nil {
 		return time.Time{}, fmt.Errorf("cannot parse %s: %w", s, err)
 	}

app/vminsert/common/streamaggr.go

@@ -36,9 +36,6 @@ var (
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
 	streamAggrIgnoreFirstIntervals = flag.Int("streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
-	streamAggrEnableWindows = flag.Bool("streamAggr.enableWindows", false, "Enables aggregation within fixed windows for all aggregators. "+
-		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
-		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 )
 
 var (
@@ -65,7 +62,6 @@ func CheckStreamAggrConfig() error {
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
-		EnableWindows:        *streamAggrEnableWindows,
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushNoop, opts, "global")
 	if err != nil {
@@ -82,7 +78,7 @@ func InitStreamAggr() {
 	saCfgReloaderStopCh = make(chan struct{})
 	if *streamAggrConfig == "" {
 		if *streamAggrDedupInterval > 0 {
-			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrEnableWindows, *streamAggrDedupInterval, *streamAggrDropInputLabels, "global")
+			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrDedupInterval, *streamAggrDropInputLabels, "global")
 		}
 		return
 	}

app/vminsert/graphite/request_handler.go

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

app/vminsert/influx/request_handler.go

@@ -34,7 +34,7 @@ var (
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
 func InsertHandlerForReader(r io.Reader) error {
-	return stream.Parse(r, "", true, "", "", func(db string, rows []parser.Row) error {
+	return stream.Parse(r, true, false, "", "", func(db string, rows []parser.Row) error {
 		return insertRows(db, rows, nil)
 	})
 }
@@ -47,13 +47,13 @@ func InsertHandlerForHTTP(req *http.Request) error {
 	if err != nil {
 		return err
 	}
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	isStreamMode := req.Header.Get("Stream-Mode") == "1"
 	q := req.URL.Query()
 	precision := q.Get("precision")
 	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
 	db := q.Get("db")
-	encoding := req.Header.Get("Content-Encoding")
-	isStreamMode := req.Header.Get("Stream-Mode") == "1"
-	return stream.Parse(req.Body, encoding, isStreamMode, precision, db, func(db string, rows []parser.Row) error {
+	return stream.Parse(req.Body, isStreamMode, isGzipped, precision, db, func(db string, rows []parser.Row) error {
 		return insertRows(db, rows, extraLabels)
 	})
 }

app/vminsert/native/request_handler.go

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

app/vminsert/newrelic/request_handler.go

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

app/vminsert/opentelemetry/request_handler.go

@@ -24,7 +24,7 @@ func InsertHandler(req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
 	var processBody func([]byte) ([]byte, error)
 	if req.Header.Get("Content-Type") == "application/json" {
 		if req.Header.Get("X-Amz-Firehose-Protocol-Version") != "" {
@@ -33,7 +33,7 @@ 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, isGzipped, processBody, func(tss []prompbmarshal.TimeSeries) error {
 		return insertRows(tss, extraLabels)
 	})
 }

app/vminsert/prometheusimport/request_handler.go

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

app/vminsert/prompush/push.go

@@ -2,7 +2,6 @@ package prompush
 
 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/metrics"
@@ -50,7 +49,6 @@ func push(ctx *common.InsertCtx, tss []prompbmarshal.TimeSeries) {
 	}
 	ctx.Reset(rowsLen)
 	rowsTotal := 0
-	hasRelabeling := relabel.HasRelabeling()
 	for i := range tss {
 		ts := &tss[i]
 		rowsTotal += len(ts.Samples)
@@ -59,7 +57,7 @@ func push(ctx *common.InsertCtx, tss []prompbmarshal.TimeSeries) {
 			label := &ts.Labels[j]
 			ctx.AddLabel(label.Name, label.Value)
 		}
-		if !ctx.TryPrepareLabels(hasRelabeling) {
+		if !ctx.TryPrepareLabels(false) {
 			continue
 		}
 		var metricNameRaw []byte

app/vminsert/vmimport/request_handler.go

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

app/vmselect/graphite/eval.go

@@ -64,7 +64,7 @@ type series struct {
 
 	expr graphiteql.Expr
 
-	// consolidateFunc is applied to raw samples in order to generate data points aligned to the given step.
+	// consolidateFunc is applied to raw samples in order to generate data points algined to the given step.
 	// see series.consolidate() function for details.
 	consolidateFunc aggrFunc
 

app/vmselect/graphite/eval_test.go

@@ -36,7 +36,7 @@ func TestExecExprSuccess(t *testing.T) {
 		if err := compareSeries(ss, expectedSeries, expr); err != nil {
 			t.Fatalf("series mismatch for query %q: %s\ngot series\n%s\nexpected series\n%s", query, err, printSeriess(ss), printSeriess(expectedSeries))
 		}
-		// make sure ec isn't changed during query execution.
+		// make sure ec isn't changed during query exection.
 		if !reflect.DeepEqual(ec, &ecCopy) {
 			t.Fatalf("unexpected ec\ngot\n%v\nwant\n%v", &ecCopy, ec)
 		}

app/vmselect/graphite/functions.json

@@ -159,7 +159,7 @@
   "asPercent": {
     "name": "asPercent",
     "function": "asPercent(seriesList, total=None, *nodes)",
-    "description": "Calculates a percentage of the total of a wildcard series. If `total` is specified,\neach series will be calculated as a percentage of that total. If `total` is not specified,\nthe sum of all points in the wildcard series will be used instead.\n\nA list of nodes can optionally be provided, if so they will be used to match series with their\ncorresponding totals following the same logic as :py:func:`groupByNodes <groupByNodes>`.\n\nWhen passing `nodes` the `total` parameter may be a series list or `None`.  If it is `None` then\nfor each series in `seriesList` the percentage of the sum of series in that group will be returned.\n\nWhen not passing `nodes`, the `total` parameter may be a single series, reference the same number\nof series as `seriesList` or be a numeric value.\n\nExample:\n\n.. code-block:: none\n\n  # Server01 connections failed and succeeded as a percentage of Server01 connections attempted\n  &target=asPercent(Server01.connections.{failed,succeeded}, Server01.connections.attempted)\n\n  # For each server, its connections failed as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.failed, Server*.connections.attempted)\n\n  # For each server, its connections failed and succeeded as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.{failed,succeeded}, Server*.connections.attempted, 0)\n\n  # apache01.threads.busy as a percentage of 1500\n  &target=asPercent(apache01.threads.busy,1500)\n\n  # Server01 cpu stats as a percentage of its total\n  &target=asPercent(Server01.cpu.*.jiffies)\n\n  # cpu stats for each server as a percentage of its total\n  &target=asPercent(Server*.cpu.*.jiffies, None, 0)\n\nWhen using `nodes`, any series or totals that can't be matched will create output series with\nnames like ``asPercent(someSeries,MISSING)`` or ``asPercent(MISSING,someTotalSeries)`` and all\nvalues set to None. If desired these series can be filtered out by piping the result through\n``|exclude(\"MISSING\")`` as shown below:\n\n.. code-block:: none\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)\n\n  # will produce 3 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n  # asPercent(Server2.memory.used,MISSING) [all values will be None]\n  # asPercent(MISSING,Server3.memory.total) [all values will be None]\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)|exclude(\"MISSING\")\n\n  # will produce 1 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n\nEach node may be an integer referencing a node in the series name or a string identifying a tag.\n\n.. note::\n\n  When `total` is a seriesList, specifying `nodes` to match series with the corresponding total\n  series will increase reliability.",
+    "description": "Calculates a percentage of the total of a wildcard series. If `total` is specified,\neach series will be calculated as a percentage of that total. If `total` is not specified,\nthe sum of all points in the wildcard series will be used instead.\n\nA list of nodes can optionally be provided, if so they will be used to match series with their\ncorresponding totals following the same logic as :py:func:`groupByNodes <groupByNodes>`.\n\nWhen passing `nodes` the `total` parameter may be a series list or `None`.  If it is `None` then\nfor each series in `seriesList` the percentage of the sum of series in that group will be returned.\n\nWhen not passing `nodes`, the `total` parameter may be a single series, reference the same number\nof series as `seriesList` or be a numeric value.\n\nExample:\n\n.. code-block:: none\n\n  # Server01 connections failed and succeeded as a percentage of Server01 connections attempted\n  &target=asPercent(Server01.connections.{failed,succeeded}, Server01.connections.attempted)\n\n  # For each server, its connections failed as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.failed, Server*.connections.attempted)\n\n  # For each server, its connections failed and succeeded as a percentage of its connections attemped\n  &target=asPercent(Server*.connections.{failed,succeeded}, Server*.connections.attempted, 0)\n\n  # apache01.threads.busy as a percentage of 1500\n  &target=asPercent(apache01.threads.busy,1500)\n\n  # Server01 cpu stats as a percentage of its total\n  &target=asPercent(Server01.cpu.*.jiffies)\n\n  # cpu stats for each server as a percentage of its total\n  &target=asPercent(Server*.cpu.*.jiffies, None, 0)\n\nWhen using `nodes`, any series or totals that can't be matched will create output series with\nnames like ``asPercent(someSeries,MISSING)`` or ``asPercent(MISSING,someTotalSeries)`` and all\nvalues set to None. If desired these series can be filtered out by piping the result through\n``|exclude(\"MISSING\")`` as shown below:\n\n.. code-block:: none\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)\n\n  # will produce 3 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n  # asPercent(Server2.memory.used,MISSING) [all values will be None]\n  # asPercent(MISSING,Server3.memory.total) [all values will be None]\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)|exclude(\"MISSING\")\n\n  # will produce 1 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n\nEach node may be an integer referencing a node in the series name or a string identifying a tag.\n\n.. note::\n\n  When `total` is a seriesList, specifying `nodes` to match series with the corresponding total\n  series will increase reliability.",
     "module": "graphite.render.functions",
     "group": "Combine",
     "params": [
@@ -635,7 +635,7 @@
   "pct": {
     "name": "pct",
     "function": "pct(seriesList, total=None, *nodes)",
-    "description": "Calculates a percentage of the total of a wildcard series. If `total` is specified,\neach series will be calculated as a percentage of that total. If `total` is not specified,\nthe sum of all points in the wildcard series will be used instead.\n\nA list of nodes can optionally be provided, if so they will be used to match series with their\ncorresponding totals following the same logic as :py:func:`groupByNodes <groupByNodes>`.\n\nWhen passing `nodes` the `total` parameter may be a series list or `None`.  If it is `None` then\nfor each series in `seriesList` the percentage of the sum of series in that group will be returned.\n\nWhen not passing `nodes`, the `total` parameter may be a single series, reference the same number\nof series as `seriesList` or be a numeric value.\n\nExample:\n\n.. code-block:: none\n\n  # Server01 connections failed and succeeded as a percentage of Server01 connections attempted\n  &target=asPercent(Server01.connections.{failed,succeeded}, Server01.connections.attempted)\n\n  # For each server, its connections failed as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.failed, Server*.connections.attempted)\n\n  # For each server, its connections failed and succeeded as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.{failed,succeeded}, Server*.connections.attempted, 0)\n\n  # apache01.threads.busy as a percentage of 1500\n  &target=asPercent(apache01.threads.busy,1500)\n\n  # Server01 cpu stats as a percentage of its total\n  &target=asPercent(Server01.cpu.*.jiffies)\n\n  # cpu stats for each server as a percentage of its total\n  &target=asPercent(Server*.cpu.*.jiffies, None, 0)\n\nWhen using `nodes`, any series or totals that can't be matched will create output series with\nnames like ``asPercent(someSeries,MISSING)`` or ``asPercent(MISSING,someTotalSeries)`` and all\nvalues set to None. If desired these series can be filtered out by piping the result through\n``|exclude(\"MISSING\")`` as shown below:\n\n.. code-block:: none\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)\n\n  # will produce 3 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n  # asPercent(Server2.memory.used,MISSING) [all values will be None]\n  # asPercent(MISSING,Server3.memory.total) [all values will be None]\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)|exclude(\"MISSING\")\n\n  # will produce 1 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n\nEach node may be an integer referencing a node in the series name or a string identifying a tag.\n\n.. note::\n\n  When `total` is a seriesList, specifying `nodes` to match series with the corresponding total\n  series will increase reliability.",
+    "description": "Calculates a percentage of the total of a wildcard series. If `total` is specified,\neach series will be calculated as a percentage of that total. If `total` is not specified,\nthe sum of all points in the wildcard series will be used instead.\n\nA list of nodes can optionally be provided, if so they will be used to match series with their\ncorresponding totals following the same logic as :py:func:`groupByNodes <groupByNodes>`.\n\nWhen passing `nodes` the `total` parameter may be a series list or `None`.  If it is `None` then\nfor each series in `seriesList` the percentage of the sum of series in that group will be returned.\n\nWhen not passing `nodes`, the `total` parameter may be a single series, reference the same number\nof series as `seriesList` or be a numeric value.\n\nExample:\n\n.. code-block:: none\n\n  # Server01 connections failed and succeeded as a percentage of Server01 connections attempted\n  &target=asPercent(Server01.connections.{failed,succeeded}, Server01.connections.attempted)\n\n  # For each server, its connections failed as a percentage of its connections attempted\n  &target=asPercent(Server*.connections.failed, Server*.connections.attempted)\n\n  # For each server, its connections failed and succeeded as a percentage of its connections attemped\n  &target=asPercent(Server*.connections.{failed,succeeded}, Server*.connections.attempted, 0)\n\n  # apache01.threads.busy as a percentage of 1500\n  &target=asPercent(apache01.threads.busy,1500)\n\n  # Server01 cpu stats as a percentage of its total\n  &target=asPercent(Server01.cpu.*.jiffies)\n\n  # cpu stats for each server as a percentage of its total\n  &target=asPercent(Server*.cpu.*.jiffies, None, 0)\n\nWhen using `nodes`, any series or totals that can't be matched will create output series with\nnames like ``asPercent(someSeries,MISSING)`` or ``asPercent(MISSING,someTotalSeries)`` and all\nvalues set to None. If desired these series can be filtered out by piping the result through\n``|exclude(\"MISSING\")`` as shown below:\n\n.. code-block:: none\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)\n\n  # will produce 3 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n  # asPercent(Server2.memory.used,MISSING) [all values will be None]\n  # asPercent(MISSING,Server3.memory.total) [all values will be None]\n\n  &target=asPercent(Server{1,2}.memory.used,Server{1,3}.memory.total,0)|exclude(\"MISSING\")\n\n  # will produce 1 output series:\n  # asPercent(Server1.memory.used,Server1.memory.total) [values will be as expected]\n\nEach node may be an integer referencing a node in the series name or a string identifying a tag.\n\n.. note::\n\n  When `total` is a seriesList, specifying `nodes` to match series with the corresponding total\n  series will increase reliability.",
     "module": "graphite.render.functions",
     "group": "Combine",
     "params": [
@@ -988,7 +988,7 @@
   "integralByInterval": {
     "name": "integralByInterval",
     "function": "integralByInterval(seriesList, intervalUnit)",
-    "description": "This will do the same as integral() function, except resetting the total to 0\nat the given time in the parameter \"from\"\nUseful for finding totals per hour/day/week/..\n\nExample:\n\n.. code-block:: none\n\n  &target=integralByInterval(company.sales.perMinute, \"1d\")&from=midnight-10days\n\nThis would start at zero on the left side of the graph, adding the sales each\nminute, and show the evolution of sales per day during the last 10 days.",
+    "description": "This will do the same as integral() funcion, except resetting the total to 0\nat the given time in the parameter \"from\"\nUseful for finding totals per hour/day/week/..\n\nExample:\n\n.. code-block:: none\n\n  &target=integralByInterval(company.sales.perMinute, \"1d\")&from=midnight-10days\n\nThis would start at zero on the left side of the graph, adding the sales each\nminute, and show the evolution of sales per day during the last 10 days.",
     "module": "graphite.render.functions",
     "group": "Transform",
     "params": [
@@ -1776,7 +1776,7 @@
   "movingAverage": {
     "name": "movingAverage",
     "function": "movingAverage(seriesList, windowSize, xFilesFactor=None)",
-    "description": "Graphs the moving average of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\naverage of the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingAverage(Server.instance01.threads.busy,10)\n  &target=movingAverage(Server.instance*.threads.idle,'5min')",
+    "description": "Graphs the moving average of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\naverage of the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingAverage(Server.instance01.threads.busy,10)\n  &target=movingAverage(Server.instance*.threads.idle,'5min')",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [
@@ -1809,7 +1809,7 @@
   "movingMax": {
     "name": "movingMax",
     "function": "movingMax(seriesList, windowSize, xFilesFactor=None)",
-    "description": "Graphs the moving maximum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nmaximum of the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMax(Server.instance01.requests,10)\n  &target=movingMax(Server.instance*.errors,'5min')",
+    "description": "Graphs the moving maximum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nmaximum of the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMax(Server.instance01.requests,10)\n  &target=movingMax(Server.instance*.errors,'5min')",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [
@@ -1842,7 +1842,7 @@
   "movingMedian": {
     "name": "movingMedian",
     "function": "movingMedian(seriesList, windowSize, xFilesFactor=None)",
-    "description": "Graphs the moving median of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nmedian of the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMedian(Server.instance01.threads.busy,10)\n  &target=movingMedian(Server.instance*.threads.idle,'5min')",
+    "description": "Graphs the moving median of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nmedian of the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMedian(Server.instance01.threads.busy,10)\n  &target=movingMedian(Server.instance*.threads.idle,'5min')",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [
@@ -1875,7 +1875,7 @@
   "movingMin": {
     "name": "movingMin",
     "function": "movingMin(seriesList, windowSize, xFilesFactor=None)",
-    "description": "Graphs the moving minimum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nminimum of the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMin(Server.instance01.requests,10)\n  &target=movingMin(Server.instance*.errors,'5min')",
+    "description": "Graphs the moving minimum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nminimum of the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingMin(Server.instance01.requests,10)\n  &target=movingMin(Server.instance*.errors,'5min')",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [
@@ -1908,7 +1908,7 @@
   "movingSum": {
     "name": "movingSum",
     "function": "movingSum(seriesList, windowSize, xFilesFactor=None)",
-    "description": "Graphs the moving sum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nsum of the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingSum(Server.instance01.requests,10)\n  &target=movingSum(Server.instance*.errors,'5min')",
+    "description": "Graphs the moving sum of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList followed by a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), and an xFilesFactor value to specify\nhow many points in the window must be non-null for the output to be considered valid. Graphs the\nsum of the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingSum(Server.instance01.requests,10)\n  &target=movingSum(Server.instance*.errors,'5min')",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [
@@ -1941,7 +1941,7 @@
   "movingWindow": {
     "name": "movingWindow",
     "function": "movingWindow(seriesList, windowSize, func='average', xFilesFactor=None)",
-    "description": "Graphs a moving window function of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList, a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), a function to apply to the points\nin the window to produce the output, and an xFilesFactor value to specify how many points in the\nwindow must be non-null for the output to be considered valid. Graphs the\noutput of the function for the preceding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingWindow(Server.instance01.threads.busy,10)\n  &target=movingWindow(Server.instance*.threads.idle,'5min','median',0.5)\n\n.. note::\n\n  `xFilesFactor` follows the same semantics as in Whisper storage schemas.  Setting it to 0 (the\n  default) means that only a single value in a given interval needs to be non-null, setting it to\n  1 means that all values in the interval must be non-null.  A setting of 0.5 means that at least\n  half the values in the interval must be non-null.",
+    "description": "Graphs a moving window function of a metric (or metrics) over a fixed number of\npast points, or a time interval.\n\nTakes one metric or a wildcard seriesList, a number N of datapoints\nor a quoted string with a length of time like '1hour' or '5min' (See ``from /\nuntil`` in the :doc:`Render API <render_api>` for examples of time formats), a function to apply to the points\nin the window to produce the output, and an xFilesFactor value to specify how many points in the\nwindow must be non-null for the output to be considered valid. Graphs the\noutput of the function for the preceeding datapoints for each point on the graph.\n\nExample:\n\n.. code-block:: none\n\n  &target=movingWindow(Server.instance01.threads.busy,10)\n  &target=movingWindow(Server.instance*.threads.idle,'5min','median',0.5)\n\n.. note::\n\n  `xFilesFactor` follows the same semantics as in Whisper storage schemas.  Setting it to 0 (the\n  default) means that only a single value in a given interval needs to be non-null, setting it to\n  1 means that all values in the interval must be non-null.  A setting of 0.5 means that at least\n  half the values in the interval must be non-null.",
     "module": "graphite.render.functions",
     "group": "Calculate",
     "params": [

app/vmselect/main.go

@@ -15,7 +15,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/stats"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
@@ -30,10 +29,7 @@ import (
 )
 
 var (
-	deleteAuthKey                = flagutil.NewPassword("deleteAuthKey", "authKey for metrics' deletion via /api/v1/admin/tsdb/delete_series and /tags/delSeries. It could be passed via authKey query arg. It overrides -httpAuth.*")
-	metricNamesStatsResetAuthKey = flagutil.NewPassword("metricNamesStatsResetAuthKey", "authKey for resetting metric names usage cache via /api/v1/admin/status/metric_names_stats/reset. It overrides -httpAuth.*. "+
-		"See https://docs.victoriametrics.com/#track-ingested-metrics-usage")
-
+	deleteAuthKey         = flagutil.NewPassword("deleteAuthKey", "authKey for metrics' deletion via /api/v1/admin/tsdb/delete_series and /tags/delSeries. It could be passed via authKey query arg. It overrides -httpAuth.*")
 	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
 		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
 		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
@@ -182,6 +178,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		promql.ResetRollupResultCache()
 		return true
 	}
+
 	if strings.HasPrefix(path, "/api/v1/label/") {
 		s := path[len("/api/v1/label/"):]
 		if strings.HasSuffix(s, "/values") {
@@ -402,26 +399,6 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		}
 		w.WriteHeader(http.StatusNoContent)
 		return true
-	case "/api/v1/status/metric_names_stats":
-		metricNamesStatsRequests.Inc()
-		if err := stats.MetricNamesStatsHandler(qt, w, r); err != nil {
-			metricNamesStatsErrors.Inc()
-			httpserver.Errorf(w, r, "%s", err)
-			return true
-		}
-		return true
-	case "/api/v1/admin/status/metric_names_stats/reset":
-		metricNamesStatsResetRequests.Inc()
-		if !httpserver.CheckAuthFlag(w, r, metricNamesStatsResetAuthKey) {
-			return true
-		}
-		if err := stats.ResetMetricNamesStatsHandler(qt); err != nil {
-			metricNamesStatsResetErrors.Inc()
-			httpserver.Errorf(w, r, "%s", err)
-			return true
-		}
-		w.WriteHeader(http.StatusNoContent)
-		return true
 	default:
 		return false
 	}
@@ -697,12 +674,6 @@ var (
 	metadataRequests       = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/metadata"}`)
 	buildInfoRequests      = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/buildinfo"}`)
 	queryExemplarsRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/query_exemplars"}`)
-
-	metricNamesStatsRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/status/metric_names_stats"}`)
-	metricNamesStatsErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/api/v1/status/metric_names_stats"}`)
-
-	metricNamesStatsResetRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/admin/status/metric_names_stats/reset"}`)
-	metricNamesStatsResetErrors   = metrics.NewCounter(`vm_http_request_errors_total{path="/api/v1/admin/status/metric_names_stats/reset"}`)
 )
 
 func proxyVMAlertRequests(w http.ResponseWriter, r *http.Request) {