app/vmselect: log calls to /api/v1/admin/tsdb/delete_series

The log message will display when deletion API was called, how many series
it deleted and what params were used. This should help identifying events
of metrics deletion.

Example:
```
2026-06-12T13:02:28.006Z        info    VictoriaMetrics/app/vmselect/prometheus/prometheus.go:529       /api/v1/admin/tsdb/delete_series has been called for "[{__name__=\"vm_http_request_errors_total\"}]". Deleted 0 series.
```

Signed-off-by: hagen1778 <roman@victoriametrics.com>

.github/workflows/build.yml

@@ -33,7 +33,8 @@ jobs:
     name: ${{ matrix.os }}-${{ matrix.arch }}
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources to increase build speed.
+    runs-on: 'vm-runner'
     strategy:
       fail-fast: false
       matrix:

.github/workflows/test.yml

@@ -30,7 +30,8 @@ jobs:
     name: lint
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources since golangci-lint requires extra memory
+    runs-on: 'vm-runner'
     steps:
       - name: Code checkout
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
@@ -64,7 +65,8 @@ jobs:
     name: unit
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources to increase tests speed.
+    runs-on: 'vm-runner'
 
     strategy:
       matrix:
@@ -95,6 +97,7 @@ jobs:
     name: apptest
     permissions:
       contents: read
+    # Runs on dedicated runner to isolate app tests from other tests.
     runs-on: apptest
 
     steps:

.golangci.yml

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

Makefile

@@ -17,7 +17,7 @@ EXTRA_GO_BUILD_TAGS ?=
 GO_BUILDINFO = -X '$(PKG_PREFIX)/lib/buildinfo.Version=$(APP_NAME)-$(DATEINFO_TAG)-$(BUILDINFO_TAG)'
 TAR_OWNERSHIP ?= --owner=1000 --group=1000
 
-GOLANGCI_LINT_VERSION := 2.9.0
+GOLANGCI_LINT_VERSION := 2.12.2
 
 .PHONY: $(MAKECMDGOALS)
 
@@ -527,7 +527,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run --build-tags 'synctest'
 
 install-golangci-lint:
-	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
+	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://golangci-lint.run/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`

app/victoria-metrics/test/prom_writer.go

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

app/vmalert-tool/unittest/input.go

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

app/vmalert-tool/unittest/recording.go

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

app/vmalert-tool/unittest/unittest.go

@@ -329,11 +329,11 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 
 	q, err := datasource.Init(nil)
 	if err != nil {
-		return []error{fmt.Errorf("failed to init datasource: %v", err)}
+		return []error{fmt.Errorf("failed to init datasource: %w", err)}
 	}
 	rw, err := remotewrite.NewDebugClient()
 	if err != nil {
-		return []error{fmt.Errorf("failed to init wr: %v", err)}
+		return []error{fmt.Errorf("failed to init wr: %w", err)}
 	}
 
 	alertEvalTimesMap := map[time.Duration]struct{}{}

app/vmalert/config/config.go

@@ -173,9 +173,9 @@ func (r *Rule) String() string {
 	if r.Alert != "" {
 		ruleType = "alerting"
 	}
-	b := strings.Builder{}
-	b.WriteString(fmt.Sprintf("%s rule %q", ruleType, r.Name()))
-	b.WriteString(fmt.Sprintf("; expr: %q", r.Expr))
+	var b strings.Builder
+	fmt.Fprintf(&b, "%s rule %q", ruleType, r.Name())
+	fmt.Fprintf(&b, "; expr: %q", r.Expr)
 
 	kv := sortMap(r.Labels)
 	for i := range kv {

app/vmalert/datasource/client_prom.go

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

app/vmalert/rule/alerting.go

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

app/vmalert/web.go

@@ -77,7 +77,7 @@ var (
 func marshalJson(v any, kind string) ([]byte, *httpserver.ErrorWithStatusCode) {
 	data, err := json.Marshal(v)
 	if err != nil {
-		return nil, errResponse(fmt.Errorf("failed to marshal %s: %s", kind, err), http.StatusInternalServerError)
+		return nil, errResponse(fmt.Errorf("failed to marshal %s: %w", kind, err), http.StatusInternalServerError)
 	}
 	return data, nil
 }

app/vmauth/main_test.go

@@ -1639,7 +1639,7 @@ func (w *fakeResponseWriter) WriteHeader(statusCode int) {
 		"X-Content-Type-Options": true,
 	})
 	if err != nil {
-		panic(fmt.Errorf("cannot marshal headers: %s", err))
+		panic(fmt.Errorf("cannot marshal headers: %w", err))
 	}
 }
 

app/vmauth/oidc.go

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

app/vmctl/influx.go

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

app/vmctl/influx/influx.go

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

app/vmctl/influx/parser.go

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

app/vmctl/main.go

@@ -88,7 +88,7 @@ func main() {
 
 					tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_opentsdb")
 					if err != nil {
-						return fmt.Errorf("failed to create transport for -%s=%q: %s", otsdbAddr, addr, err)
+						return fmt.Errorf("failed to create transport for -%s=%q: %w", otsdbAddr, addr, err)
 					}
 					oCfg := opentsdb.Config{
 						Addr:       addr,
@@ -103,17 +103,17 @@ func main() {
 					}
 					otsdbClient, err := opentsdb.NewClient(oCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create opentsdb client: %s", err)
+						return fmt.Errorf("failed to create opentsdb client: %w", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 
 					otsdbProcessor := newOtsdbProcessor(otsdbClient, importer, c.Int(otsdbConcurrency), c.Bool(globalVerbose))
@@ -137,7 +137,7 @@ func main() {
 
 					tc, err := promauth.NewTLSConfig(certFile, keyFile, caFile, serverName, insecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %s", err)
+						return fmt.Errorf("failed to create TLS Config: %w", err)
 					}
 
 					iCfg := influx.Config{
@@ -157,17 +157,17 @@ func main() {
 
 					influxClient, err := influx.NewClient(iCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create influx client: %s", err)
+						return fmt.Errorf("failed to create influx client: %w", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 
 					processor := newInfluxProcessor(
@@ -203,7 +203,7 @@ func main() {
 
 					tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_remoteread")
 					if err != nil {
-						return fmt.Errorf("failed to create transport for -%s=%q: %s", remoteReadSrcAddr, addr, err)
+						return fmt.Errorf("failed to create transport for -%s=%q: %w", remoteReadSrcAddr, addr, err)
 					}
 
 					// Backwards compatible default values if none provided by user
@@ -227,17 +227,17 @@ func main() {
 						DisablePathAppend: c.Bool(remoteReadDisablePathAppend),
 					})
 					if err != nil {
-						return fmt.Errorf("error create remote read client: %s", err)
+						return fmt.Errorf("error create remote read client: %w", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 
 					rmp := remoteReadProcessor{
@@ -265,12 +265,12 @@ func main() {
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 
 					promCfg := prometheus.Config{
@@ -285,7 +285,7 @@ func main() {
 					}
 					cl, err := prometheus.NewClient(promCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create prometheus client: %s", err)
+						return fmt.Errorf("failed to create prometheus client: %w", err)
 					}
 
 					pp := prometheusProcessor{
@@ -307,12 +307,12 @@ func main() {
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 
 					mCfg := mimir.Config{
@@ -335,7 +335,7 @@ func main() {
 					}
 					cl, err := mimir.NewClient(ctx, mCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create mimir client: %s", err)
+						return fmt.Errorf("failed to create mimir client: %w", err)
 					}
 
 					pp := prometheusProcessor{
@@ -356,12 +356,12 @@ func main() {
 					fmt.Println("Thanos import mode")
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %s", err)
+						return fmt.Errorf("failed to init VM configuration: %w", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %s", err)
+						return fmt.Errorf("failed to create VM importer: %w", err)
 					}
 					thanosCfg := thanos.Config{
 						Snapshot: c.String(thanosSnapshot),
@@ -374,7 +374,7 @@ func main() {
 					}
 					cl, err := thanos.NewClient(thanosCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create thanos client: %s", err)
+						return fmt.Errorf("failed to create thanos client: %w", err)
 					}
 
 					var aggrTypes []thanos.AggrType
@@ -382,7 +382,7 @@ func main() {
 						for _, typeStr := range aggrTypesStr {
 							aggrType, err := thanos.ParseAggrType(typeStr)
 							if err != nil {
-								return fmt.Errorf("failed to parse aggregate type %q: %s", typeStr, err)
+								return fmt.Errorf("failed to parse aggregate type %q: %w", typeStr, err)
 							}
 							aggrTypes = append(aggrTypes, aggrType)
 						}
@@ -415,7 +415,7 @@ func main() {
 					bfMinDuration := c.Duration(vmNativeBackoffMinDuration)
 					bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
 					if err != nil {
-						return fmt.Errorf("failed to create backoff object: %s", err)
+						return fmt.Errorf("failed to create backoff object: %w", err)
 					}
 
 					disableKeepAlive := c.Bool(vmNativeDisableHTTPKeepAlive)
@@ -439,7 +439,7 @@ func main() {
 
 					srcTC, err := promauth.NewTLSConfig(srcCertFile, srcKeyFile, srcCAFile, srcServerName, srcInsecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %s", err)
+						return fmt.Errorf("failed to create TLS Config: %w", err)
 					}
 
 					trSrc := httputil.NewTransport(false, "vmctl_src")
@@ -469,7 +469,7 @@ func main() {
 
 					dstTC, err := promauth.NewTLSConfig(dstCertFile, dstKeyFile, dstCAFile, dstServerName, dstInsecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %s", err)
+						return fmt.Errorf("failed to create TLS Config: %w", err)
 					}
 
 					trDst := httputil.NewTransport(false, "vmctl_dst")
@@ -534,7 +534,7 @@ func main() {
 					log.Printf("verifying block at path=%q", blockPath)
 					f, err := os.OpenFile(blockPath, os.O_RDONLY, 0600)
 					if err != nil {
-						return cli.Exit(fmt.Errorf("cannot open exported block at path=%q err=%w", blockPath, err), 1)
+						return cli.Exit(fmt.Errorf("cannot open exported block at path=%q: %w", blockPath, err), 1)
 					}
 					defer f.Close()
 					var blocksCount atomic.Uint64
@@ -542,7 +542,7 @@ func main() {
 						blocksCount.Add(1)
 						return nil
 					}); err != nil {
-						return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d, err=%w", blockPath, blocksCount.Load(), err), 1)
+						return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d: %w", blockPath, blocksCount.Load(), err), 1)
 					}
 					log.Printf("successfully verified block at path=%q, blockCount=%d", blockPath, blocksCount.Load())
 					return nil
@@ -585,7 +585,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
 
 	tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_client")
 	if err != nil {
-		return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %s", vmAddr, addr, err)
+		return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %w", vmAddr, addr, err)
 	}
 
 	bfRetries := c.Int(vmBackoffRetries)
@@ -593,7 +593,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
 	bfMinDuration := c.Duration(vmBackoffMinDuration)
 	bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
 	if err != nil {
-		return vm.Config{}, fmt.Errorf("failed to create backoff object: %s", err)
+		return vm.Config{}, fmt.Errorf("failed to create backoff object: %w", err)
 	}
 
 	return vm.Config{

app/vmctl/mimir/lazyreader.go

@@ -54,7 +54,7 @@ func (lbr *lazyBlockReader) initialize() error {
 	// fetching block and parse it and store it in lbr.reader
 	temp, err := lbr.mkTempDir()
 	if err != nil {
-		return fmt.Errorf("failed to create temp dir: %s", err)
+		return fmt.Errorf("failed to create temp dir: %w", err)
 	}
 
 	lbr.tempDirPath = temp
@@ -85,7 +85,7 @@ func (lbr *lazyBlockReader) initialize() error {
 			return fmt.Errorf("failed to fetch chunk file: %q: %w", chunkName, err)
 		}
 		if err := lbr.writeFile(temp, blockChunkPath, chunk); err != nil {
-			return fmt.Errorf("failed to write chunk file: %q: %s", chunkName, err)
+			return fmt.Errorf("failed to write chunk file: %q: %w", chunkName, err)
 		}
 	}
 
@@ -135,7 +135,7 @@ func (lbr *lazyBlockReader) Meta() tsdb.BlockMeta {
 // Size returns the number of bytes that the block takes up on disk.
 func (lbr *lazyBlockReader) Size() int64 {
 	if err := lbr.initialize(); err != nil {
-		lbr.err = fmt.Errorf("error get Size of the block: %s, return zero size", err)
+		lbr.err = fmt.Errorf("error get Size of the block: %w, return zero size", err)
 		return 0
 	}
 	return lbr.reader.Size()
@@ -167,11 +167,11 @@ func (lbr *lazyBlockReader) Close() error {
 func (lbr *lazyBlockReader) mkTempDir() (string, error) {
 	temp, err := os.MkdirTemp("", lbr.ID.String())
 	if err != nil {
-		return "", fmt.Errorf("failed to create temp dir: %s", err)
+		return "", fmt.Errorf("failed to create temp dir: %w", err)
 	}
 	err = os.Mkdir(filepath.Join(temp, "chunks"), os.ModePerm)
 	if err != nil {
-		return "", fmt.Errorf("failed to create temp dir: %s", err)
+		return "", fmt.Errorf("failed to create temp dir: %w", err)
 	}
 	return temp, nil
 }

app/vmctl/mimir/mimir.go

@@ -133,11 +133,11 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 	c.RemoteFS = rfs
 	timeMin, err := utils.ParseTime(cfg.Filter.TimeMin)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse min time in filter: %s", err)
+		return nil, fmt.Errorf("failed to parse min time in filter: %w", err)
 	}
 	timeMax, err := utils.ParseTime(cfg.Filter.TimeMax)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse max time in filter: %s", err)
+		return nil, fmt.Errorf("failed to parse max time in filter: %w", err)
 	}
 	c.filter = filter{
 		min:        timeMin.UnixMilli(),
@@ -156,7 +156,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 
 	indexFile, err := c.fetchIndexFile()
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch index file: %s", err)
+		return nil, fmt.Errorf("failed to fetch index file: %w", err)
 	}
 
 	var blocksToImport []tsdb.BlockReader
@@ -172,7 +172,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 
 		lazyBlockReader, err := newLazyBlockReader(block, c.RemoteFS)
 		if err != nil {
-			return nil, fmt.Errorf("failed to create lazy block reader: %s", err)
+			return nil, fmt.Errorf("failed to create lazy block reader: %w", err)
 		}
 		blocksToImport = append(blocksToImport, lazyBlockReader)
 	}
@@ -185,7 +185,7 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 func (c *Client) Read(ctx context.Context, block tsdb.BlockReader) (*prometheus.CloseableSeriesSet, error) {
 	meta := block.Meta()
 	if b, ok := block.(*lazyBlockReader); ok && b.Err() != nil {
-		return nil, fmt.Errorf("failed to read block: %s", b.Err())
+		return nil, fmt.Errorf("failed to read block: %w", b.Err())
 	}
 
 	if meta.ULID.String() == "" {
@@ -218,20 +218,20 @@ func (c *Client) fetchIndexFile() (*Index, error) {
 
 	file, err := c.ReadFile(bucketIndexCompressedFilename)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read bucket index: %s", err)
+		return nil, fmt.Errorf("failed to read bucket index: %w", err)
 	}
 
 	r := bytes.NewReader(file)
 	// Read all the content.
 	gzipReader, err := gzip.NewReader(r)
 	if err != nil {
-		return nil, fmt.Errorf("failed to create gzip reader: %s", err)
+		return nil, fmt.Errorf("failed to create gzip reader: %w", err)
 	}
 
 	var indexFile Index
 	err = json.NewDecoder(gzipReader).Decode(&indexFile)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decode bucket index: %s", err)
+		return nil, fmt.Errorf("failed to decode bucket index: %w", err)
 	}
 
 	return &indexFile, nil

app/vmctl/native/client.go

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

app/vmctl/opentsdb.go

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

app/vmctl/opentsdb/opentsdb.go

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

app/vmctl/opentsdb/parser.go

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

app/vmctl/prometheus.go

@@ -46,7 +46,7 @@ type prometheusProcessor struct {
 func (pp *prometheusProcessor) run(ctx context.Context) error {
 	blocks, err := pp.cl.Explore()
 	if err != nil {
-		return fmt.Errorf("explore failed: %s", err)
+		return fmt.Errorf("explore failed: %w", err)
 	}
 	if len(blocks) < 1 {
 		return fmt.Errorf("found no blocks to import")
@@ -57,7 +57,7 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
 	}
 
 	if err := pp.processBlocks(ctx, blocks); err != nil {
-		return fmt.Errorf("migration failed: %s", err)
+		return fmt.Errorf("migration failed: %w", err)
 	}
 
 	log.Println("Import finished!")
@@ -68,7 +68,7 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
 func (pp *prometheusProcessor) do(ctx context.Context, b tsdb.BlockReader) error {
 	css, err := pp.cl.Read(ctx, b)
 	if err != nil {
-		return fmt.Errorf("failed to read block: %s", err)
+		return fmt.Errorf("failed to read block: %w", err)
 	}
 	defer func() {
 		if err := css.Close(); err != nil {
@@ -146,7 +146,7 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 			for br := range blockReadersCh {
 				if err := pp.do(ctx, br); err != nil {
 					promErrorsTotal.Inc()
-					errCh <- fmt.Errorf("cannot read block %q: %s", br.Meta().ULID, err)
+					errCh <- fmt.Errorf("cannot read block %q: %w", br.Meta().ULID, err)
 					return
 				}
 				if cb, ok := br.(io.Closer); ok {
@@ -164,11 +164,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 		select {
 		case promErr := <-errCh:
 			close(blockReadersCh)
-			return fmt.Errorf("prometheus error: %s", promErr)
+			return fmt.Errorf("prometheus error: %w", promErr)
 		case vmErr := <-pp.im.Errors():
 			close(blockReadersCh)
 			promErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
+			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
 		case blockReadersCh <- br:
 		}
 	}
@@ -182,11 +182,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 	for vmErr := range pp.im.Errors() {
 		if vmErr.Err != nil {
 			promErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
+			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
 		}
 	}
 	for err := range errCh {
-		return fmt.Errorf("import process failed: %s", err)
+		return fmt.Errorf("import process failed: %w", err)
 	}
 
 	return nil

app/vmctl/prometheus/prometheus.go

@@ -59,12 +59,12 @@ func (f filter) inRange(minV, maxV int64) bool {
 func NewClient(cfg Config) (*Client, error) {
 	db, err := tsdb.OpenDBReadOnly(cfg.Snapshot, cfg.TemporaryDir, nil)
 	if err != nil {
-		return nil, fmt.Errorf("failed to open snapshot %q: %s", cfg.Snapshot, err)
+		return nil, fmt.Errorf("failed to open snapshot %q: %w", cfg.Snapshot, err)
 	}
 	c := &Client{DBReadOnly: db}
 	timeMin, timeMax, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse time in filter: %s", err)
+		return nil, fmt.Errorf("failed to parse time in filter: %w", err)
 	}
 	c.filter = filter{
 		min:        timeMin,
@@ -83,7 +83,7 @@ func NewClient(cfg Config) (*Client, error) {
 func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 	blocks, err := c.Blocks()
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch blocks: %s", err)
+		return nil, fmt.Errorf("failed to fetch blocks: %w", err)
 	}
 	s := &vmctlutil.Stats{
 		Filtered: c.filter.min != 0 || c.filter.max != 0 || c.filter.label != "",
@@ -142,14 +142,14 @@ func parseTime(start, end string) (int64, int64, error) {
 	if start != "" {
 		v, err := time.Parse(time.RFC3339, start)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %s", start, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %w", start, err)
 		}
 		s = v.UnixNano() / int64(time.Millisecond)
 	}
 	if end != "" {
 		v, err := time.Parse(time.RFC3339, end)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %s", end, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %w", end, err)
 		}
 		e = v.UnixNano() / int64(time.Millisecond)
 	}

app/vmctl/remoteread.go

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

app/vmctl/remoteread/remoteread.go

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

app/vmctl/thanos/client.go

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

app/vmctl/thanos_processor.go

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

app/vmctl/vm/vm.go

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

app/vmctl/vm_native.go

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

app/vmselect/prometheus/prometheus.go

@@ -28,6 +28,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -525,6 +526,7 @@ func DeleteHandler(startTime time.Time, r *http.Request) error {
 	if deletedCount > 0 {
 		promql.ResetRollupResultCache()
 	}
+	logger.Infof("/api/v1/admin/tsdb/delete_series has been called for %q. Deleted %d series.", sq.FiltersString(), deletedCount)
 	return nil
 }
 

app/vmstorage/main.go

@@ -10,8 +10,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -23,6 +21,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vminsertapi"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
+	"github.com/VictoriaMetrics/metrics"
 )
 
 var (
@@ -153,7 +152,7 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 		LogNewSeries:                *logNewSeries,
 	}
 	strg := storage.MustOpenStorage(*storageDataPath, opts)
-	vmStorage = newVMStorageSingleNode(strg, vmselectMaxConcurrentRequests, resetCacheIfNeeded)
+	vmStorage = newVMStorage(strg, vmselectMaxConcurrentRequests, resetCacheIfNeeded)
 
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)
@@ -175,15 +174,15 @@ func Init(vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []stora
 	GetSearch = vmStorage.GetSearch
 	PutSearch = vmStorage.PutSearch
 	RequestHandler = vmStorage.requestHandler
-	DebugFlush = vmStorage.vms.s.DebugFlush
+	DebugFlush = vmStorage.s.DebugFlush
 }
 
 var storageMetrics *metrics.Set
 
 var (
-	// vmStorageSingleNode is an instance of vmstorage used by vminsert and
+	// vmStorage is an instance of vmstorage used by vminsert and
 	// vmselect for writing and reading data.
-	vmStorage      *VMStorageSingleNode
+	vmStorage      *VMStorage
 	VMInsertAPI    vminsertapi.API
 	VMSelectAPI    vmselectapi.API
 	GetSearch      func(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error)
@@ -209,15 +208,12 @@ func Stop() {
 	logger.Infof("the vmstorage has been stopped")
 }
 
-func (vmssn *VMStorageSingleNode) requestHandler(w http.ResponseWriter, r *http.Request) bool {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.requestHandler(w, r)
-}
-
 // requestHandler is a storage request handler.
 // TODO(@rtm0): Move to a separate file, request_handler.go
 func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) bool {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	path := r.URL.Path
 	if path == "/internal/force_merge" {
 		if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
@@ -257,7 +253,7 @@ func (vms *VMStorage) requestHandler(w http.ResponseWriter, r *http.Request) boo
 			dealine, err = strconv.Atoi(deadlineStr)
 			if err != nil {
 				logger.Errorf("cannot parse `seconds` arg %q: %s", deadlineStr, err)
-				jsonResponseError(w, fmt.Errorf("cannot parse `seconds` arg %q: %s", deadlineStr, err))
+				jsonResponseError(w, fmt.Errorf("cannot parse `seconds` arg %q: %w", deadlineStr, err))
 				return true
 			}
 		}
@@ -361,15 +357,11 @@ var (
 	snapshotsDeleteAllErrorsTotal = metrics.NewCounter(`vm_http_request_errors_total{path="/snapshot/delete_all"}`)
 )
 
-// TODO(@rtm0): Move to metrics.go.
-func (vmssn *VMStorageSingleNode) writeStorageMetrics(w io.Writer) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	vmssn.vms.writeStorageMetrics(w)
-}
-
 // TODO(@rtm0): Move to metrics.go.
 func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	strg := vms.s
 	var m storage.Metrics
 	strg.UpdateMetrics(&m)

app/vmstorage/vmstorage.go

@@ -1,6 +1,7 @@
 package vmstorage
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"sync"
@@ -14,6 +15,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
 )
@@ -34,7 +36,7 @@ var (
 // newVMStorage creates a new instance of of VMStorage.
 //
 // The created VMStorage instance takes ownership of s.
-func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int) *VMStorage {
+func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) *VMStorage {
 	if err := encoding.CheckPrecisionBits(uint8(*precisionBits)); err != nil {
 		logger.Fatalf("invalid -precisionBits=%d: %s", *precisionBits, err)
 	}
@@ -49,6 +51,8 @@ func newVMStorage(s *storage.Storage, vmselectMaxConcurrentRequests int) *VMStor
 		maxUniqueTimeseries:           *maxUniqueTimeseries,
 		maxUniqueTimeSeriesCalculated: maxUniqueTimeseriesCalculated,
 		staleSnapshotsRemoverCh:       make(chan struct{}),
+		wg:                            syncwg.WaitGroup{},
+		resetCacheIfNeeded:            resetCacheIfNeeded,
 	}
 	vms.initStaleSnapshotsRemover()
 	return vms
@@ -78,6 +82,17 @@ type VMStorage struct {
 	maxUniqueTimeSeriesCalculated int
 	staleSnapshotsRemoverCh       chan struct{}
 	staleSnapshotsRemoverWG       sync.WaitGroup
+
+	// wg is used to wrap every storage call into wg.Add(1) ... wg.Done()
+	// for proper graceful shutdown when Stop is called.
+	//
+	// Use syncwg instead of sync, since Add is called from concurrent
+	// goroutines.
+	wg syncwg.WaitGroup
+
+	// resetCacheIfNeeded is a callback for automatic resetting of response
+	// cache if needed.
+	resetCacheIfNeeded func(mrs []storage.MetricRow)
 }
 
 func (vms *VMStorage) initStaleSnapshotsRemover() {
@@ -103,6 +118,7 @@ func (vms *VMStorage) initStaleSnapshotsRemover() {
 func (vms *VMStorage) Stop() {
 	close(vms.staleSnapshotsRemoverCh)
 	vms.staleSnapshotsRemoverWG.Wait()
+	vms.wg.WaitAndBlock()
 	vms.s.MustClose()
 }
 
@@ -111,6 +127,14 @@ func (vms *VMStorage) Stop() {
 // The caller should limit the number of concurrent calls to WriteRows() in
 // order to limit memory usage.
 func (vms *VMStorage) WriteRows(rows []storage.MetricRow) error {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
+	if vms.s.IsReadOnly() {
+		return errReadOnly
+	}
+	vms.resetCacheIfNeeded(rows)
+
 	vms.s.AddRows(rows, uint8(*precisionBits))
 	return nil
 }
@@ -120,26 +144,41 @@ func (vms *VMStorage) WriteRows(rows []storage.MetricRow) error {
 // The caller should limit the number of concurrent calls to WriteMetadata() in
 // order to limit memory usage.
 func (vms *VMStorage) WriteMetadata(rows []metricsmetadata.Row) error {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
+	if vms.s.IsReadOnly() {
+		return errReadOnly
+	}
 	vms.s.AddMetadataRows(rows)
 	return nil
 }
 
+var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
+
 // IsReadOnly returns true is the storage is in read-only mode.
 func (vms *VMStorage) IsReadOnly() bool {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
 	return vms.s.IsReadOnly()
 }
 
 func (vms *VMStorage) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
+	vms.wg.Add(1)
+
 	tr := sq.GetTimeRange()
 	maxMetrics := vms.getMaxMetrics(sq.MaxMetrics)
 	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
 	if err != nil {
+		vms.wg.Done()
 		return nil, err
 	}
 	if len(tfss) == 0 {
+		vms.wg.Done()
 		return nil, fmt.Errorf("missing tag filters")
 	}
 	bi := getBlockIterator()
+	bi.wgDone = vms.wg.Done
 	bi.sr.Init(qt, vms.s, tfss, tr, maxMetrics, deadline)
 	if err := bi.sr.Error(); err != nil {
 		bi.MustClose()
@@ -161,8 +200,9 @@ func (vms *VMStorage) getMaxMetrics(searchQueryLimit int) int {
 
 // blockIterator implements vmselectapi.BlockIterator
 type blockIterator struct {
-	sr storage.Search
-	mb storage.MetricBlock
+	sr     storage.Search
+	mb     storage.MetricBlock
+	wgDone func()
 }
 
 var blockIteratorsPool sync.Pool
@@ -171,6 +211,8 @@ func (bi *blockIterator) MustClose() {
 	bi.sr.MustClose()
 	bi.mb.MetricName = nil
 	bi.mb.Block.Reset()
+	bi.wgDone()
+	bi.wgDone = nil
 	blockIteratorsPool.Put(bi)
 }
 
@@ -197,8 +239,63 @@ func (bi *blockIterator) Error() error {
 	return bi.sr.Error()
 }
 
+// GetSearch sets up an instance of storage search and returns it to the caller
+// along with the max series count that the search can return.
+//
+// This method is not part of the vmselectapi.API and must only be used by
+// vmsingle HTTP handlers.
+//
+// Callers of this method must call PutSearch() once the search instance is not
+// needed anymore.
+func (vms *VMStorage) GetSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error) {
+	vms.wg.Add(1)
+
+	tr := sq.GetTimeRange()
+	maxMetrics := vms.getMaxMetrics(sq.MaxMetrics)
+	tfss, err := vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
+	if err != nil {
+		vms.wg.Done()
+		return nil, 0, err
+	}
+
+	sr := getSearch()
+	maxSeriesCount := sr.Init(qt, vms.s, tfss, tr, sq.MaxMetrics, deadline)
+	return sr, maxSeriesCount, nil
+}
+
+// PutSearch resets the search once it is not needed anymore and puts it aside
+// for future reuse.
+//
+// This method is not part of the vmselectapi.API and must only be used by
+// vmsingle HTTP handlers.
+//
+// The method must only be used on search instances that have been created with
+// GetSearch().
+func (vms *VMStorage) PutSearch(sr *storage.Search) {
+	putSearch(sr)
+	vms.wg.Done()
+}
+
+func getSearch() *storage.Search {
+	v := ssPool.Get()
+	if v == nil {
+		return &storage.Search{}
+	}
+	return v.(*storage.Search)
+}
+
+func putSearch(sr *storage.Search) {
+	sr.MustClose()
+	ssPool.Put(sr)
+}
+
+var ssPool sync.Pool
+
 // SearchMetricNames returns metric names for the given tfss on the given tr.
 func (vms *VMStorage) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -219,6 +316,9 @@ func (vms *VMStorage) SearchMetricNames(qt *querytracer.Tracer, sq *storage.Sear
 // SearchLabelValues searches for label values for the given labelName, tfss and
 // tr.
 func (vms *VMStorage) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	tr := sq.GetTimeRange()
 	if maxLabelValues <= 0 || maxLabelValues > *maxTagValues {
 		maxLabelValues = *maxTagValues
@@ -244,6 +344,9 @@ func (vms *VMStorage) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuer
 // similar APIs.
 func (vms *VMStorage) TagValueSuffixes(qt *querytracer.Tracer, _, _ uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte,
 	maxSuffixes int, deadline uint64) ([]string, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	if maxSuffixes <= 0 || maxSuffixes > *maxTagValueSuffixesPerSearch {
 		maxSuffixes = *maxTagValueSuffixesPerSearch
 	}
@@ -260,6 +363,9 @@ func (vms *VMStorage) TagValueSuffixes(qt *querytracer.Tracer, _, _ uint32, tr s
 
 // SearchLabelNames searches for tag keys matching the given tfss on tr.
 func (vms *VMStorage) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	tr := sq.GetTimeRange()
 	if maxLabelNames <= 0 || maxLabelNames > *maxTagKeys {
 		maxLabelNames = *maxTagKeys
@@ -278,6 +384,8 @@ func (vms *VMStorage) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery
 }
 
 func (vms *VMStorage) SeriesCount(_ *querytracer.Tracer, _, _ uint32, deadline uint64) (uint64, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
 	return vms.s.GetSeriesCount(deadline)
 }
 
@@ -287,6 +395,9 @@ func (vms *VMStorage) Tenants(_ *querytracer.Tracer, _ storage.TimeRange, _ uint
 
 // GetTSDBStatus returns TSDB status for given filters on the given date.
 func (vms *VMStorage) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -306,6 +417,9 @@ func (vms *VMStorage) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery
 //
 // Returns the number of deleted series.
 func (vms *VMStorage) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	tr := sq.GetTimeRange()
 	maxMetrics := sq.MaxMetrics
 	if maxMetrics <= 0 {
@@ -324,17 +438,26 @@ func (vms *VMStorage) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQue
 }
 
 func (vms *VMStorage) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, _ uint64) error {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	vms.s.RegisterMetricNames(qt, mrs)
 	return nil
 }
 
 // GetMetricNamesUsageStats returns metric name usage stats.
 func (vms *VMStorage) GetMetricNamesUsageStats(qt *querytracer.Tracer, _ *storage.TenantToken, limit, le int, matchPattern string, _ uint64) (metricnamestats.StatsResult, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	return vms.s.GetMetricNamesStats(qt, limit, le, matchPattern), nil
 }
 
 // ResetMetricNamesStats resets state for metric names usage tracker
 func (vms *VMStorage) ResetMetricNamesUsageStats(qt *querytracer.Tracer, _ uint64) error {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
+
 	vms.s.ResetMetricNamesStats(qt)
 	return nil
 }
@@ -371,6 +494,8 @@ func (vms *VMStorage) setupTfss(qt *querytracer.Tracer, sq *storage.SearchQuery,
 }
 
 func (vms *VMStorage) GetMetadataRecords(qt *querytracer.Tracer, _ *storage.TenantToken, limit int, metricName string, _ uint64) ([]*metricsmetadata.Row, error) {
+	vms.wg.Add(1)
+	defer vms.wg.Done()
 	return vms.s.GetMetadataRows(qt, limit, metricName), nil
 }
 

app/vmstorage/vmstorage_single_node.go

@@ -1,213 +0,0 @@
-package vmstorage
-
-import (
-	"errors"
-	"fmt"
-	"sync"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricsmetadata"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/syncwg"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmselectapi"
-)
-
-// newVMStorageSingleNode creates a new instance of of VMStorage for vmsingle.
-func newVMStorageSingleNode(s *storage.Storage, maxConcurrentRequests int, resetCacheIfNeeded func(mrs []storage.MetricRow)) *VMStorageSingleNode {
-	vms := newVMStorage(s, maxConcurrentRequests)
-	return &VMStorageSingleNode{
-		vms:                vms,
-		wg:                 syncwg.WaitGroup{},
-		resetCacheIfNeeded: resetCacheIfNeeded,
-	}
-}
-
-type VMStorageSingleNode struct {
-	vms *VMStorage
-
-	// wg is used to wrap every storage call into wg.Add(1) ... wg.Done()
-	// for proper graceful shutdown when Stop is called.
-	//
-	// Use syncwg instead of sync, since Add is called from concurrent
-	// goroutines.
-	wg syncwg.WaitGroup
-
-	// resetCacheIfNeeded is a callback for automatic resetting of response
-	// cache if needed.
-	resetCacheIfNeeded func(mrs []storage.MetricRow)
-}
-
-func (vmssn *VMStorageSingleNode) Stop() {
-	vmssn.wg.WaitAndBlock()
-	vmssn.vms.Stop()
-}
-
-// WriteRows writes metric rows to the storage.
-//
-// Returns an error if the storage is in read-only mode.
-//
-// The caller should limit the number of concurrent calls to WriteRows() in
-// order to limit memory usage.
-func (vmssn *VMStorageSingleNode) WriteRows(rows []storage.MetricRow) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-
-	if vmssn.vms.IsReadOnly() {
-		return errReadOnly
-	}
-	vmssn.resetCacheIfNeeded(rows)
-	return vmssn.vms.WriteRows(rows)
-}
-
-// WriteMetadata writes metrics metadata to storage.
-//
-// Returns an error if the storage is in read-only mode.
-//
-// The caller should limit the number of concurrent calls to WriteMetadata() in
-// order to limit memory usage.
-func (vmssn *VMStorageSingleNode) WriteMetadata(rows []metricsmetadata.Row) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-
-	if vmssn.vms.IsReadOnly() {
-		return errReadOnly
-	}
-	return vmssn.vms.WriteMetadata(rows)
-}
-
-var errReadOnly = errors.New("the storage is in read-only mode; check -storage.minFreeDiskSpaceBytes command-line flag value")
-
-func (vmssn *VMStorageSingleNode) IsReadOnly() bool {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.IsReadOnly()
-}
-
-func (vmssn *VMStorageSingleNode) InitSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (vmselectapi.BlockIterator, error) {
-	return nil, fmt.Errorf("not implemented in vmsingle")
-}
-
-// GetSearch sets up an instance of storage search and returns it to the caller
-// along with the max series count that the search can return.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// Callers of this method must call PutSearch() once the search instance is not
-// needed anymore.
-func (vmssn *VMStorageSingleNode) GetSearch(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (*storage.Search, int, error) {
-	vmssn.wg.Add(1)
-
-	tr := sq.GetTimeRange()
-	maxMetrics := vmssn.vms.getMaxMetrics(sq.MaxMetrics)
-	tfss, err := vmssn.vms.setupTfss(qt, sq, tr, maxMetrics, deadline)
-	if err != nil {
-		vmssn.wg.Done()
-		return nil, 0, err
-	}
-
-	sr := getSearch()
-	maxSeriesCount := sr.Init(qt, vmssn.vms.s, tfss, tr, sq.MaxMetrics, deadline)
-	return sr, maxSeriesCount, nil
-}
-
-// PutSearch resets the search once it is not needed anymore and puts it aside
-// for future reuse.
-//
-// This method is not part of the vmselectapi.API and must only be used by
-// vmsingle HTTP handlers.
-//
-// The method must only be used on search instances that have been created with
-// GetSearch().
-func (vmssn *VMStorageSingleNode) PutSearch(sr *storage.Search) {
-	putSearch(sr)
-	vmssn.wg.Done()
-}
-
-func getSearch() *storage.Search {
-	v := ssPool.Get()
-	if v == nil {
-		return &storage.Search{}
-	}
-	return v.(*storage.Search)
-}
-
-func putSearch(sr *storage.Search) {
-	sr.MustClose()
-	ssPool.Put(sr)
-}
-
-var ssPool sync.Pool
-
-func (vmssn *VMStorageSingleNode) SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.SearchMetricNames(qt, sq, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) LabelValues(qt *querytracer.Tracer, sq *storage.SearchQuery, labelName string, maxLabelValues int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.LabelValues(qt, sq, labelName, maxLabelValues, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) TagValueSuffixes(qt *querytracer.Tracer, accountID, projectID uint32, tr storage.TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxSuffixes int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.TagValueSuffixes(qt, accountID, projectID, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames int, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.LabelNames(qt, sq, maxLabelNames, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) SeriesCount(qt *querytracer.Tracer, accountID, projectID uint32, deadline uint64) (uint64, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.SeriesCount(qt, accountID, projectID, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) Tenants(qt *querytracer.Tracer, tr storage.TimeRange, deadline uint64) ([]string, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.Tenants(qt, tr, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel string, topN int, deadline uint64) (*storage.TSDBStatus, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.TSDBStatus(qt, sq, focusLabel, topN, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline uint64) (int, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.DeleteSeries(qt, sq, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) RegisterMetricNames(qt *querytracer.Tracer, mrs []storage.MetricRow, deadline uint64) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.RegisterMetricNames(qt, mrs, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) GetMetricNamesUsageStats(qt *querytracer.Tracer, tt *storage.TenantToken, limit, le int, matchPattern string, deadline uint64) (metricnamestats.StatsResult, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.GetMetricNamesUsageStats(qt, tt, limit, le, matchPattern, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) ResetMetricNamesUsageStats(qt *querytracer.Tracer, deadline uint64) error {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.ResetMetricNamesUsageStats(qt, deadline)
-}
-
-func (vmssn *VMStorageSingleNode) GetMetadataRecords(qt *querytracer.Tracer, tt *storage.TenantToken, limit int, metricName string, deadline uint64) ([]*metricsmetadata.Row, error) {
-	vmssn.wg.Add(1)
-	defer vmssn.wg.Done()
-	return vmssn.vms.GetMetadataRecords(qt, tt, limit, metricName, deadline)
-}

app/vmstorage/vmstorage_test.go

@@ -48,7 +48,7 @@ func TestGetMaxMetrics(t *testing.T) {
 		t.Helper()
 		*maxUniqueTimeseries = storageMaxUniqueTimeseries
 		s := storage.MustOpenStorage(t.Name(), storage.OpenOptions{})
-		vms := newVMStorage(s, maxConcurrentRequests)
+		vms := newVMStorage(s, maxConcurrentRequests, func(mrs []storage.MetricRow) {})
 		defer vms.Stop()
 		maxMetrics := vms.getMaxMetrics(searchQueryLimit)
 		if maxMetrics != expect {

app/vmui/packages/vmui/package.json

@@ -21,16 +21,16 @@
   },
   "dependencies": {
     "classnames": "^2.5.1",
-    "dayjs": "^1.11.20",
+    "dayjs": "^1.11.21",
     "lodash.debounce": "^4.0.8",
-    "marked": "^18.0.2",
-    "preact": "^10.29.1",
-    "qs": "^6.15.1",
+    "marked": "^18.0.5",
+    "preact": "^10.29.2",
+    "qs": "^6.15.2",
     "react-input-mask": "^2.0.4",
-    "react-router-dom": "^7.14.1",
+    "react-router-dom": "^7.17.0",
     "uplot": "^1.6.32",
-    "vite": "^8.0.8",
-    "web-vitals": "^5.2.0"
+    "vite": "^8.0.16",
+    "web-vitals": "^5.3.0"
   },
   "devDependencies": {
     "@eslint/eslintrc": "^3.3.5",
@@ -39,24 +39,24 @@
     "@testing-library/jest-dom": "^6.9.1",
     "@testing-library/preact": "^3.2.4",
     "@types/lodash.debounce": "^4.0.9",
-    "@types/node": "^25.6.0",
-    "@types/qs": "^6.15.0",
-    "@types/react": "^19.2.14",
+    "@types/node": "^25.9.2",
+    "@types/qs": "^6.15.1",
+    "@types/react": "^19.2.17",
     "@types/react-input-mask": "^3.0.6",
     "@types/react-router-dom": "^5.3.3",
-    "@typescript-eslint/eslint-plugin": "^8.58.2",
-    "@typescript-eslint/parser": "^8.58.2",
+    "@typescript-eslint/eslint-plugin": "^8.61.0",
+    "@typescript-eslint/parser": "^8.61.0",
     "cross-env": "^10.1.0",
     "eslint": "^9.39.2",
     "eslint-plugin-react": "^7.37.5",
     "eslint-plugin-unused-imports": "^4.4.1",
-    "globals": "^17.5.0",
-    "http-proxy-middleware": "^3.0.5",
-    "jsdom": "^29.0.2",
-    "postcss": "^8.5.10",
-    "sass-embedded": "^1.99.0",
-    "typescript": "^6.0.2",
-    "vitest": "^4.1.4"
+    "globals": "^17.6.0",
+    "http-proxy-middleware": "^4.1.0",
+    "jsdom": "^29.1.1",
+    "postcss": "^8.5.15",
+    "sass-embedded": "^1.100.0",
+    "typescript": "^6.0.3",
+    "vitest": "^4.1.8"
   },
   "browserslist": {
     "production": [

apptest/client.go

@@ -156,14 +156,14 @@ func readAllAndClose(t *testing.T, responseBody io.ReadCloser) string {
 //
 // This type is expected to be embedded by the apps that serve metrics.
 type metricsClient struct {
-	metricsCli *Client
-	url        string
+	cli *Client
+	url string
 }
 
 func newMetricsClient(cli *Client, addr string) *metricsClient {
 	return &metricsClient{
-		metricsCli: cli,
-		url:        fmt.Sprintf("http://%s/metrics", addr),
+		cli: cli,
+		url: fmt.Sprintf("http://%s/metrics", addr),
 	}
 }
 
@@ -179,7 +179,7 @@ func (c *metricsClient) GetIntMetric(t *testing.T, metricName string) int {
 func (c *metricsClient) GetMetric(t *testing.T, metricName string) float64 {
 	t.Helper()
 
-	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
+	metrics, statusCode := c.cli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -205,7 +205,7 @@ func (c *metricsClient) GetMetricsByPrefix(t *testing.T, prefix string) []float6
 
 	values := []float64{}
 
-	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
+	metrics, statusCode := c.cli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -234,7 +234,7 @@ func (c *metricsClient) GetMetricsByRegexp(t *testing.T, re *regexp.Regexp) []fl
 
 	values := []float64{}
 
-	metrics, statusCode := c.metricsCli.Get(t, c.url, nil)
+	metrics, statusCode := c.cli.Get(t, c.url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -270,7 +270,7 @@ func (c *metricsClient) rpcRowsSentTotal(t *testing.T) int {
 }
 
 type vmselectClient struct {
-	vmselectCli              *Client
+	cli                      *Client
 	url                      func(op, path string, opts QueryOpts) string
 	metricNamesStatsResetURL string
 	tenantsURL               string
@@ -287,7 +287,7 @@ func (c *vmselectClient) PrometheusAPIV1Export(t *testing.T, query string, opts
 	values := opts.asURLValues()
 	values.Add("match[]", query)
 	values.Add("format", "promapi")
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -302,7 +302,7 @@ func (c *vmselectClient) PrometheusAPIV1ExportNative(t *testing.T, query string,
 	values := opts.asURLValues()
 	values.Add("match[]", query)
 	values.Add("format", "promapi")
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return []byte(res)
 }
 
@@ -315,7 +315,7 @@ func (c *vmselectClient) PrometheusAPIV1Query(t *testing.T, query string, opts Q
 	url := c.url("select", "prometheus/api/v1/query", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -329,7 +329,7 @@ func (c *vmselectClient) PrometheusAPIV1QueryRange(t *testing.T, query string, o
 	url := c.url("select", "prometheus/api/v1/query_range", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1QueryResponse(t, res)
 }
 
@@ -342,7 +342,7 @@ func (c *vmselectClient) PrometheusAPIV1Series(t *testing.T, matchQuery string,
 	url := c.url("select", "prometheus/api/v1/series", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1SeriesResponse(t, res)
 }
 
@@ -354,7 +354,7 @@ func (c *vmselectClient) PrometheusAPIV1SeriesCount(t *testing.T, opts QueryOpts
 	t.Helper()
 	url := c.url("select", "prometheus/api/v1/series/count", opts)
 	values := opts.asURLValues()
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1SeriesCountResponse(t, res)
 }
 
@@ -367,7 +367,7 @@ func (c *vmselectClient) PrometheusAPIV1Labels(t *testing.T, matchQuery string,
 	url := c.url("select", "prometheus/api/v1/labels", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1LabelsResponse(t, res)
 }
 
@@ -382,7 +382,7 @@ func (c *vmselectClient) PrometheusAPIV1LabelValues(t *testing.T, labelName, mat
 	url := c.url("select", path, opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1LabelValuesResponse(t, res)
 }
 
@@ -394,7 +394,7 @@ func (c *vmselectClient) PrometheusAPIV1Metadata(t *testing.T, metric string, li
 	values := opts.asURLValues()
 	values.Add("metric", metric)
 	values.Add("limit", strconv.Itoa(limit))
-	res, _ := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, _ := c.cli.PostForm(t, url, values, opts.Headers)
 	return NewPrometheusAPIV1Metadata(t, res)
 }
 
@@ -408,7 +408,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminTSDBDeleteSeries(t *testing.T, matc
 	url := c.url("delete", "prometheus/api/v1/admin/tsdb/delete_series", opts)
 	values := opts.asURLValues()
 	values.Add("match[]", matchQuery)
-	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusNoContent {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
 	}
@@ -426,7 +426,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusMetricNamesStats(t *testing.T, lim
 	values.Add("limit", limit)
 	values.Add("le", le)
 	values.Add("match_pattern", matchPattern)
-	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -455,7 +455,7 @@ func (c *vmselectClient) PrometheusAPIV1StatusTSDB(t *testing.T, matchQuery stri
 	addNonEmpty("match[]", matchQuery)
 	addNonEmpty("topN", topN)
 	addNonEmpty("date", date)
-	res, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	res, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -476,7 +476,7 @@ func (c *vmselectClient) GraphiteMetricsIndex(t *testing.T, opts QueryOpts) Grap
 	t.Helper()
 
 	url := c.url("select", "graphite/metrics/index.json", opts)
-	res, statusCode := c.vmselectCli.Get(t, url, opts.Headers)
+	res, statusCode := c.cli.Get(t, url, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -499,7 +499,7 @@ func (c *vmselectClient) GraphiteMetricsFind(t *testing.T, query string, opts Qu
 	url := c.url("select", "graphite/metrics/find", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -522,7 +522,7 @@ func (c *vmselectClient) GraphiteMetricsExpand(t *testing.T, query string, opts
 	url := c.url("select", "graphite/metrics/expand", opts)
 	values := opts.asURLValues()
 	values.Add("query", query)
-	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -546,7 +546,7 @@ func (c *vmselectClient) GraphiteRender(t *testing.T, target string, opts QueryO
 	values := opts.asURLValues()
 	values.Add("format", "json")
 	values.Add("target", target)
-	resText, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	resText, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, resText)
 	}
@@ -567,7 +567,7 @@ func (c *vmselectClient) GraphiteTagsTagSeries(t *testing.T, record string, opts
 	url := c.url("select", "graphite/tags/tagSeries", opts)
 	values := opts.asURLValues()
 	values.Add("path", record)
-	_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if got, want := statusCode, http.StatusNotImplemented; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d", got, want)
 	}
@@ -584,7 +584,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
 	for _, rec := range records {
 		values.Add("path", rec)
 	}
-	_, statusCode := c.vmselectCli.PostForm(t, url, values, opts.Headers)
+	_, statusCode := c.cli.PostForm(t, url, values, opts.Headers)
 	if got, want := statusCode, http.StatusNotImplemented; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d", got, want)
 	}
@@ -598,7 +598,7 @@ func (c *vmselectClient) GraphiteTagsTagMultiSeries(t *testing.T, records []stri
 func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *testing.T, opts QueryOpts) {
 	t.Helper()
 	values := opts.asURLValues()
-	res, statusCode := c.vmselectCli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
+	res, statusCode := c.cli.PostForm(t, c.metricNamesStatsResetURL, values, opts.Headers)
 	if statusCode != http.StatusNoContent {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusNoContent, res)
 	}
@@ -608,7 +608,7 @@ func (c *vmselectClient) PrometheusAPIV1AdminStatusMetricNamesStatsReset(t *test
 // /admin/tenants endpoint.
 func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminTenantsResponse {
 	t.Helper()
-	res, statusCode := c.vmselectCli.Get(t, c.tenantsURL, opts.Headers)
+	res, statusCode := c.cli.Get(t, c.tenantsURL, opts.Headers)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", statusCode, http.StatusOK, res)
 	}
@@ -622,7 +622,7 @@ func (c *vmselectClient) APIV1AdminTenants(t *testing.T, opts QueryOpts) *AdminT
 }
 
 type vminsertClient struct {
-	vminsertCli        *Client
+	cli                *Client
 	url                func(op, path string, opts QueryOpts) string
 	openTSDBURL        func(op, path string, opts QueryOpts) string
 	graphiteListenAddr string
@@ -647,7 +647,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportCSV(t *testing.T, records []string
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -671,7 +671,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportNative(t *testing.T, data []byte,
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, 1, func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -693,7 +693,7 @@ func (c *vminsertClient) PrometheusAPIV1Write(t *testing.T, wr prompb.WriteReque
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/x-protobuf")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -745,7 +745,7 @@ func (c *vminsertClient) PrometheusAPIV1ImportPrometheus(t *testing.T, records [
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -771,7 +771,7 @@ func (c *vminsertClient) InfluxWrite(t *testing.T, records []string, opts QueryO
 	headers.Set("Content-Type", "text/plain")
 	c.sendBlocking(t, len(records), func() {
 		t.Helper()
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -805,7 +805,7 @@ func (c *vminsertClient) OpentelemetryV1Metrics(t *testing.T, md otlppb.MetricsD
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/x-protobuf")
 	c.sendBlocking(t, recordsCount, func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusOK {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 		}
@@ -830,7 +830,7 @@ func (c *vminsertClient) OpenTSDBAPIPut(t *testing.T, records []string, opts Que
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/json")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusNoContent {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusNoContent)
 		}
@@ -853,7 +853,7 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
 	headers := opts.getHeaders()
 	headers.Set("Content-Type", "application/json")
 	c.sendBlocking(t, len(records), func() {
-		_, statusCode := c.vminsertCli.Post(t, url, data, headers)
+		_, statusCode := c.cli.Post(t, url, data, headers)
 		if statusCode != http.StatusOK {
 			t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 		}
@@ -867,11 +867,11 @@ func (c *vminsertClient) ZabbixConnectorHistory(t *testing.T, records []string,
 // See https://docs.victoriametrics.com/victoriametrics/integrations/graphite/#ingesting
 func (c *vminsertClient) GraphiteWrite(t *testing.T, records []string, _ QueryOpts) {
 	t.Helper()
-	c.vminsertCli.Write(t, c.graphiteListenAddr, records)
+	c.cli.Write(t, c.graphiteListenAddr, records)
 }
 
 type vmstorageClient struct {
-	vmstorageCli   *Client
+	cli            *Client
 	httpListenAddr string
 }
 
@@ -881,7 +881,7 @@ func (c *vmstorageClient) ForceFlush(t *testing.T) {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/internal/force_flush", c.httpListenAddr)
-	_, statusCode := c.vmstorageCli.Get(t, url, nil)
+	_, statusCode := c.cli.Get(t, url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -892,7 +892,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/internal/force_merge", c.httpListenAddr)
-	_, statusCode := c.vmstorageCli.Get(t, url, nil)
+	_, statusCode := c.cli.Get(t, url, nil)
 	if statusCode != http.StatusOK {
 		t.Fatalf("unexpected status code: got %d, want %d", statusCode, http.StatusOK)
 	}
@@ -905,7 +905,7 @@ func (c *vmstorageClient) ForceMerge(t *testing.T) {
 func (c *vmstorageClient) SnapshotCreate(t *testing.T) *SnapshotCreateResponse {
 	t.Helper()
 
-	data, statusCode := c.vmstorageCli.Post(t, c.SnapshotCreateURL(), nil, nil)
+	data, statusCode := c.cli.Post(t, c.SnapshotCreateURL(), nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -931,7 +931,7 @@ func (c *vmstorageClient) APIV1AdminTSDBSnapshot(t *testing.T) *APIV1AdminTSDBSn
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/api/v1/admin/tsdb/snapshot", c.httpListenAddr)
-	data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
+	data, statusCode := c.cli.Post(t, url, nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -952,7 +952,7 @@ func (c *vmstorageClient) SnapshotList(t *testing.T) *SnapshotListResponse {
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/list", c.httpListenAddr)
-	data, statusCode := c.vmstorageCli.Get(t, url, nil)
+	data, statusCode := c.cli.Get(t, url, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}
@@ -973,7 +973,7 @@ func (c *vmstorageClient) SnapshotDelete(t *testing.T, snapshotName string) *Sna
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/delete?snapshot=%s", c.httpListenAddr, snapshotName)
-	data, statusCode := c.vmstorageCli.Delete(t, url)
+	data, statusCode := c.cli.Delete(t, url)
 	wantStatusCodes := map[int]bool{
 		http.StatusOK:                  true,
 		http.StatusInternalServerError: true,
@@ -998,7 +998,7 @@ func (c *vmstorageClient) SnapshotDeleteAll(t *testing.T) *SnapshotDeleteAllResp
 	t.Helper()
 
 	url := fmt.Sprintf("http://%s/snapshot/delete_all", c.httpListenAddr)
-	data, statusCode := c.vmstorageCli.Post(t, url, nil, nil)
+	data, statusCode := c.cli.Post(t, url, nil, nil)
 	if got, want := statusCode, http.StatusOK; got != want {
 		t.Fatalf("unexpected status code: got %d, want %d, resp text=%q", got, want, data)
 	}

apptest/vminsert.go

@@ -77,7 +77,7 @@ type vminsertRuntimeValues struct {
 func newVminsert(app *app, cli *Client, rt vminsertRuntimeValues) *Vminsert {
 	metricsClient := newMetricsClient(cli, rt.httpListenAddr)
 	vminsertClient := &vminsertClient{
-		vminsertCli: cli,
+		cli: cli,
 		url: func(op, path string, opts QueryOpts) string {
 			return getClusterPath(rt.httpListenAddr, op, path, opts)
 		},

apptest/vmselect.go

@@ -48,7 +48,7 @@ func newVmselect(app *app, cli *Client, rt vmselectRuntimeValues) *Vmselect {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmselectClient: &vmselectClient{
-			vmselectCli: cli,
+			cli: cli,
 			url: func(op, path string, opts QueryOpts) string {
 				return getClusterPath(rt.httpListenAddr, op, path, opts)
 			},

apptest/vmsingle.go

@@ -58,11 +58,11 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmstorageClient: &vmstorageClient{
-			vmstorageCli:   cli,
+			cli:            cli,
 			httpListenAddr: rt.httpListenAddr,
 		},
 		vmselectClient: &vmselectClient{
-			vmselectCli: cli,
+			cli: cli,
 			url: func(op, path string, opts QueryOpts) string {
 				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
 			},
@@ -70,7 +70,7 @@ func newVmsingle(app *app, cli *Client, rt vmsingleRuntimeValues) *Vmsingle {
 			tenantsURL:               "vmsingle-does-not-serve-tenants",
 		},
 		vminsertClient: &vminsertClient{
-			vminsertCli: cli,
+			cli: cli,
 			url: func(_, path string, _ QueryOpts) string {
 				return fmt.Sprintf("http://%s/%s", rt.httpListenAddr, path)
 			},

apptest/vmstorage.go

@@ -63,7 +63,7 @@ func newVmstorage(app *app, cli *Client, rt vmstorageRuntimeValues) *Vmstorage {
 		app:           app,
 		metricsClient: newMetricsClient(cli, rt.httpListenAddr),
 		vmstorageClient: &vmstorageClient{
-			vmstorageCli:   cli,
+			cli:            cli,
 			httpListenAddr: rt.httpListenAddr,
 		},
 		storageDataPath: rt.storageDataPath,

deployment/docker/vmanomaly/vmanomaly-integration/compose.yml

@@ -59,7 +59,7 @@ services:
       - '--external.alert.source=explore?orgId=1&left=["now-1h","now","VictoriaMetrics",{"expr": },{"mode":"Metrics"},{"ui":[true,true,true,"none"]}]'
     restart: always
   vmanomaly:
-    image: victoriametrics/vmanomaly:v1.29.4
+    image: victoriametrics/vmanomaly:v1.29.5
     depends_on:
       - "victoriametrics"
     ports:

deployment/docker/vmanomaly/vmanomaly-integration/vmanomaly_guide_dashboard.json

@@ -43,7 +43,7 @@
         "content": "If you don't observe any data initially, please wait a few minutes for it to appear. \n\nUpon the first running the guide (if there is not enough node_exporter monitoring data collected in your system), you may notice a significant number of false positive anomalies found. The predictions will become more accurate with at least two weeks' (full `fit_window`) worth of data provided to vmanomaly.\n\nEach row displays information for a distinct mode. The query used for anomaly detection is `sum(rate(node_cpu_seconds_total[5m])) by (mode, instance, job)`.\n",
         "mode": "markdown"
       },
-      "pluginVersion": "10.2.1",
+      "pluginVersion": "12.2.0",
       "title": "Overview",
       "type": "text"
     },

docs/anomaly-detection/CHANGELOG.md

@@ -14,6 +14,21 @@ aliases:
 ---
 Please find the changelog for VictoriaMetrics Anomaly Detection below.
 
+## v1.29.5
+Released: 2026-06-11
+
+- UI: Updated [vmanomaly UI](https://docs.victoriametrics.com/anomaly-detection/ui/) from [v1.7.0](https://docs.victoriametrics.com/anomaly-detection/ui/#v170) to [v1.7.1](https://docs.victoriametrics.com/anomaly-detection/ui/#v171), see respective [release notes](https://docs.victoriametrics.com/anomaly-detection/ui/#v171) for details.
+
+- IMPROVEMENT: Redesigned [hot reload](https://docs.victoriametrics.com/anomaly-detection/components/#hot-reload) config change detection to content-based polling with configurable `-configCheckInterval`, improving reliability for Kubernetes ConfigMap symlink rotations and other filesystems where event delivery can be inconsistent.
+
+- IMPROVEMENT: Refined config validation errors for broken or invalid config sections, so startup and reload failures point to the affected section more clearly (e.g. YAML indentation typos).
+
+- IMPROVEMENT: Tightened config validation for [`PeriodicScheduler`](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/#periodic-scheduler) `infer_every` and [`IsolationForestModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#isolation-forest-multivariate) `contamination`, including clearer handling of missing scheduler intervals, numeric contamination strings, and invalid non-finite values.
+
+- BUGFIX: Fixed a multiprocessing startup issue with `settings.n_workers > 1` that could leave scheduled data fetch or successive inference jobs stuck and repeatedly skipped by internal scheduler.
+
+- BUGFIX: Bounded [`VmReader`](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader) and [`VLogsReader`](https://docs.victoriametrics.com/anomaly-detection/components/reader/#victorialogs-reader) data fetch and post-fetch processing waits so stalled datasource reads or multiprocessing dataframe creation no longer keep [`PeriodicScheduler`](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/#periodic-scheduler) `data_fetch` jobs running indefinitely. Previously, such stuck jobs could keep internal scheduler's `max_instances=1` slot per (scheduler, query) pair occupied, causing future data fetch, fit, or infer runs to be skipped until vmanomaly was restarted. The config validator now also warns when the configured reader timeout budget can exceed the connected scheduler interval.
+
 ## v1.29.4
 Released: 2026-05-15
 

docs/anomaly-detection/FAQ.md

@@ -423,7 +423,7 @@ services:
   # ...
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.4
+    image: victoriametrics/vmanomaly:v1.29.5
     # ...
     restart: always
     volumes:
@@ -641,7 +641,7 @@ options:
 Here’s an example of using the config splitter to divide configurations based on the `extra_filters` argument from the reader section:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.4 && docker image tag victoriametrics/vmanomaly:v1.29.4 vmanomaly
+docker pull victoriametrics/vmanomaly:v1.29.5 && docker image tag victoriametrics/vmanomaly:v1.29.5 vmanomaly
 ```
 
 ```sh

docs/anomaly-detection/Migration.md

@@ -45,8 +45,7 @@ There are 2 types of compatibility to consider when migrating in stateful mode:
 
 | Group start | Group end | Compatibility | Notes |
 |---------|--------- |------------|-------|
-| [v1.29.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1293) | Latest* | Fully Compatible | Just a placeholder for new releases |
-| [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) | [v1.29.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1293) | Fully Compatible | - |
+| [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) | [v1.29.5](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1295) | Fully Compatible | - |
 | [v1.28.7](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1287) | [v1.29.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1290) | Partially compatible* | Dumped models of class [prophet](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet) and [seasonal quantile](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-seasonal-quantile) have problems with loading to [v1.29.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1290) due to dropped `pytz` library. **Upgrading directly from v1.28.7 to [v1.29.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1291) with a fix is suggested** |
 | [v1.26.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1262) | [v1.28.7](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1287) | Fully Compatible | [v1.28.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1280) introduced [rolling](https://docs.victoriametrics.com/anomaly-detection/components/models/#rolling-models) model class drop in favor of [online](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-models) models (`rolling_quantile` and `std` models), however, it does not impact compatibility, as artifacts were not produced by default for rolling models. Also, offline `mad` and `zscore` models are redirecting to their respective online counterparts since [v1.28.4](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1284). |
 | [v1.25.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1253) | [v1.26.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1270) | Partially Compatible* | [v1.25.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1253) introduced `forecast_at` argument for base [univariate](https://docs.victoriametrics.com/anomaly-detection/components/models/#univariate-models) and `Prophet` [models](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet), however, itself remains backward-reversible from newer states like [v1.26.2](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1262), [v1.27.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1270). (All models except `isolation_forest_multivariate` class will be dropped) |

docs/anomaly-detection/QuickStart.md

@@ -37,29 +37,39 @@ The `vmanomaly` service supports a set of command-line arguments to configure it
 > Single-dashed command-line argument {{% available_from "v1.23.3" anomaly %}} format can be used, e.g. `-license.forceOffline` in addition to `--license.forceOffline`. This aligns better with other VictoriaMetrics ecosystem components. Mixing the two styles is also supported, e.g. `-license.forceOffline --loggerLevel INFO`.
 
 ```shellhelp
-usage: vmanomaly.py [-h] [--license STRING | --licenseFile PATH] [--license.forceOffline] [--loggerLevel {DEBUG,WARNING,FATAL,ERROR,INFO}] [--watch] [--dryRun] [--outputSpec PATH] config [config ...]
+usage: vmanomaly.py [-h] [--license STRING | --licenseFile PATH] [--license.forceOffline] [--loggerLevel {DEBUG,INFO,WARNING,ERROR,FATAL}] [--watch] [-configCheckInterval DURATION] [--dryRun] [--outputSpec PATH] config [config ...]
 
 VictoriaMetrics Anomaly Detection Service
 
 positional arguments:
-  config                YAML config file(s) or directories containing YAML files. Multiple files will recursively merge each other values so multiple configs can be combined. If a directory is provided,
-                        all `.yaml` files inside will be merged, without recursion. Default: vmanomaly.yaml is expected in the current directory.
+  config                YAML config file(s) or directories containing YAML files. Multiple files will recursively merge each other
+                        values so multiple configs can be combined. If a directory is provided, all `.yaml` files inside will be
+                        merged, without recursion. Default: vmanomaly.yaml is expected in the current directory.
 
 options:
   -h                    Show this help message and exit
-  --license STRING      License key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
-  --licenseFile PATH    Path to file with license key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
+  --license STRING      License key for VictoriaMetrics Enterprise. See https://victoriametrics.com/products/enterprise/trial/ to
+                        obtain a trial license.
+  --licenseFile PATH    Path to file with license key for VictoriaMetrics Enterprise. See
+                        https://victoriametrics.com/products/enterprise/trial/ to obtain a trial license.
   --license.forceOffline
-                        Whether to force offline verification for VictoriaMetrics Enterprise license key, which has been passed either via -license or via -licenseFile command-line flag. The issued
-                        license key must support offline verification feature. Contact info@victoriametrics.com if you need offline license verification.
-  --loggerLevel {DEBUG,WARNING,FATAL,ERROR,INFO}
-                        Minimum level to log. Possible values: DEBUG, INFO, WARNING, ERROR, FATAL.
-  --watch               Watch config files for changes and trigger hot reloads. Watches the specified config file or directory for modifications, deletions, or additions. Upon detecting changes,
-                        triggers config reload. If new config validation fails, continues with previous valid config and state.
-  --dryRun              Validate only: parse + merge all YAML(s) and run schema checks, then exit. Does not require a license to run. Does not expose metrics, or launch vmanomaly service(s).
+                        Whether to force offline verification for VictoriaMetrics Enterprise license key, which has been passed either
+                        via -license or via -licenseFile command-line flag. The issued license key must support offline verification
+                        feature. Contact info@victoriametrics.com if you need offline license verification.
+  --loggerLevel {DEBUG,INFO,WARNING,ERROR,FATAL}
+                        Minimum level to log. Possible values: {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'FATAL'}.
+  --watch               Watch config files for changes and trigger hot reloads. Watches the specified config file or directory for
+                        modifications, deletions, or additions. Upon detecting changes, triggers config reload. If new config
+                        validation fails, continues with previous valid config and state.
+  -configCheckInterval DURATION
+                        Interval for checking watched config files for content changes. Default: 30s.
+  --dryRun              Validate only: parse + merge all YAML(s) and run schema checks, then exit. Does not require a license to run.
+                        Does not expose metrics, or launch vmanomaly service(s).
   --outputSpec PATH     Target location of .yaml output spec.
 ```
 
+{{% available_from "v1.29.5" anomaly %}} When `--watch` is enabled, config changes are detected by fixed-interval content polling instead of filesystem event delivery. The polling frequency is controlled by `-configCheckInterval` (default: `30s`). The same option can also be passed as `--configCheckInterval`, `--config.check.interval`, `--config-check-interval`, `--config_check_interval`, or in key-value form such as `configCheckInterval=30s`.
+
 You can specify these options when running `vmanomaly` to fine-tune logging levels or handle licensing configurations, as per your requirements.
 
 ### Licensing
@@ -122,7 +132,7 @@ Below are the steps to get `vmanomaly` up and running inside a Docker container:
 1. Pull Docker image:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.4
+docker pull victoriametrics/vmanomaly:v1.29.5
 ```
 
 2. Create the license file with your license key.
@@ -142,7 +152,7 @@ docker run -it \
     -v ./license:/license \
     -v ./config.yaml:/config.yaml \
     -p 8490:8490 \
-    victoriametrics/vmanomaly:v1.29.4 \
+    victoriametrics/vmanomaly:v1.29.5 \
     /config.yaml \
     --licenseFile=/license \
     --loggerLevel=INFO \
@@ -159,7 +169,7 @@ docker run -it \
     -e VMANOMALY_DATA_DUMPS_DIR=/tmp/vmanomaly/data \
     -e VMANOMALY_MODEL_DUMPS_DIR=/tmp/vmanomaly/models \
     -p 8490:8490 \
-    victoriametrics/vmanomaly:v1.29.4 \
+    victoriametrics/vmanomaly:v1.29.5 \
     /config.yaml \
     --licenseFile=/license \
     --loggerLevel=INFO \
@@ -172,7 +182,7 @@ services:
   # ...
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.4
+    image: victoriametrics/vmanomaly:v1.29.5
     # ...
     restart: always
     volumes:

docs/anomaly-detection/UI.md

@@ -315,7 +315,7 @@ docker run -it --rm \
   -e VMANOMALY_MCP_SERVER_URL=http://mcp-vmanomaly:8081/mcp \
   -p 8080:8080 \
   -p 8490:8490 \
-  victoriametrics/vmanomaly:v1.29.4 \
+  victoriametrics/vmanomaly:v1.29.5 \
   vmanomaly_config.yaml
 ```
 
@@ -640,6 +640,17 @@ If the **results** look good and the **model configuration should be deployed in
 
 ## Changelog
 
+### v1.7.1
+Released: 2026-06-11
+
+vmanomaly version: [v1.29.5](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1295)
+
+- FEATURE: Added bulk Apply/Decline actions for [Copilot](#ai-assistance) chat suggestions.
+
+- BUGFIX: Fixed modal windows closing when the mouse is released outside the window during text selection.
+
+- BUGFIX: Fixed tooltip hover behavior so tooltips do not disappear while the cursor moves into the hover content.
+
 ### v1.7.0
 Released: 2026-05-15
 

docs/anomaly-detection/components/README.md

@@ -143,11 +143,14 @@ server:
 
 > This feature is better used in conjunction with [stateful service](https://docs.victoriametrics.com/anomaly-detection/components/settings/#state-restoration) to preserve the state of the models and schedulers between restarts and reuse what can be reused, thus avoiding unnecessary re-training of models, re-initialization of schedulers and re-reading of data.
 
-{{% available_from "v1.25.0" anomaly %}} Service supports hot reload of configuration files, which allows for automatic reloading of configurations on config files change filesystem events without the need of explicit service restart. This can be enabled via the `--watch` [CLI argument](https://docs.victoriametrics.com/anomaly-detection/quickstart/#command-line-arguments). `vmanomaly_config_reload_enabled` flag in [self-monitoring metrics](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#startup-metrics) will be set to 1 (if enabled) or 0 (if disabled).
+{{% available_from "v1.25.0" anomaly %}} Service supports hot reload of configuration files, which allows for automatic reloading of configurations on config files change without the need of explicit service restart. This can be enabled via the `--watch` [CLI argument](https://docs.victoriametrics.com/anomaly-detection/quickstart/#command-line-arguments). `vmanomaly_config_reload_enabled` flag in [self-monitoring metrics](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#startup-metrics) will be set to 1 (if enabled) or 0 (if disabled).
+
+> [!NOTE]
+> {{% deprecated_from "v1.29.5" anomaly %}} File system event-based hot reload has been deprecated in favor of content-based polling with configurable `-configCheckInterval` due to reliability issues with Kubernetes ConfigMap symlink rotations and other filesystems where event delivery can be inconsistent. If you were using file system event-based hot reload, please switch to content-based polling by enabling `--watch` flag and configuring `-configCheckInterval` as needed.
 
 ### How it works
 
-It works by watching for file system events, such as modifications, creations, or deletions of `.yml|.yaml` files in the specified directories. When a change is detected, the service will attempt to reload the configuration files, rebuild the [global config](https://docs.victoriametrics.com/anomaly-detection/scaling-vmanomaly/#global-configuration) and reinitialize the components. If the reload is successful, the `vmanomaly_config_reloads_total` metric will be incremented for `status="success"` label, otherwise it will be incremented with `status="failure"` label and a respective error message on config validation failure(s) will be logged.
+It works by checking watched `.yml|.yaml` file contents in the specified files or directories on the configured interval `-configCheckInterval` (default is `30s`) {{% available_from "v1.29.5" anomaly %}}. When a content change is detected, the service will attempt to reload the configuration files after the existing debounce window, rebuild the [global config](https://docs.victoriametrics.com/anomaly-detection/scaling-vmanomaly/#global-configuration) and reinitialize the components. If the reload is successful, the `vmanomaly_config_reloads_total` metric will be incremented for `status="success"` label, otherwise it will be incremented with `status="failure"` label and a respective error message on config validation failure(s) will be logged.
 
 > If the reload fails, the service will log an error message indicating the reason for the failure, and the **previous configuration will remain active until a successful reload occurs** to preserve the service's stability. This means that if there are errors in the new configuration, the service will continue to operate with the last valid configuration until the issues are resolved.
 

docs/anomaly-detection/components/models.md

@@ -449,9 +449,9 @@ models:
 
 > The `decay` argument works only in combination with [online models](#online-models) like [`ZScoreOnlineModel`](#online-z-score) or [`OnlineQuantileModel`](#online-seasonal-quantile).
 
-The `decay` {{% available_from "v1.23.0" anomaly %}} argument is used to control the (exponential) **decay factor** for online models, which determines how quickly the model adapts to new data. It is a float value between `0.0` and `1.0`, where:
-- `1.0` means no decay (the model treats all data equally, without giving more weight to recent data). This is the default value for backward compatibility.
-- Less than `1.0` means that the model will give more weight to recent data, effectively "forgetting" older data over time.
+The `decay` {{% available_from "v1.23.0" anomaly %}} argument is used to control the (exponential) **decay factor** for online models, which determines how quickly the model adapts to new data. It is a positive float value from `(0.0, 1.0]` interval, where:
+- Value `1.0` means no decay (the model treats all data points equally, without giving more weight to recent ones). This is the default value for backward compatibility.
+- Values less than `1.0` mean that the model will give more weight to recent data, effectively "forgetting" older data over time.
 
 Roughly speaking, for the recent N datapoints model processes `decay` = `d` means that these datapoints will contribute to the model as [1 - d^X] percent of total importance, for example decay of
 - `0.99` means that 100 recent datapoints will contribute as [1 - 0.99^100] = 63.23% of total importance
@@ -998,7 +998,7 @@ Here we use Isolation Forest implementation from `scikit-learn` [library](https:
 
 * `class` (string) - model class name `"model.isolation_forest.IsolationForestMultivariateModel"` (or `isolation_forest_multivariate` with class alias support {{% available_from "v1.13.0" anomaly %}})
 
-* `contamination` (float or string, optional) - The amount of contamination of the data set, i.e. the proportion of outliers in the data set. Used when fitting to define the threshold on the scores of the samples. Default value - "auto". Should be either `"auto"` or be in the range (0.0, 0.5].
+* `contamination` (float or string, optional) - The amount of contamination of the data set, i.e. the proportion of outliers in the data set. Used when fitting to define the threshold on the scores of the samples. Default value - "auto". Should be either `"auto"` or be in the range (0.0, 0.5]. {{% available_from "v1.29.5" anomaly %}} Numeric strings, such as `"0.01"`, are accepted, while invalid non-finite values, such as `nan`, `inf`, and `-inf`, are rejected during config validation.
 
 * `seasonal_features` (list of string) - List of seasonality to encode through [cyclical encoding](https://towardsdatascience.com/cyclical-features-encoding-its-about-time-ce23581845ca), i.e. `dow` (day of week). **Introduced in [1.12.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1120)**. 
   - Empty by default for backward compatibility.
@@ -1265,7 +1265,7 @@ monitoring:
 Let's pull the docker image for `vmanomaly`:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.29.4
+docker pull victoriametrics/vmanomaly:v1.29.5
 ```
 
 Now we can run the docker container putting as volumes both config and model file:
@@ -1279,7 +1279,7 @@ docker run -it \
 -v $(PWD)/license:/license \
 -v $(PWD)/custom_model.py:/vmanomaly/model/custom.py \
 -v $(PWD)/custom.yaml:/config.yaml \
-victoriametrics/vmanomaly:v1.29.4 /config.yaml \
+victoriametrics/vmanomaly:v1.29.5 /config.yaml \
 --licenseFile=/license
 --watch
 ```

docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md

@@ -10,12 +10,12 @@ sitemap:
 
 - To use *vmanomaly*, part of the enterprise package, a license key is required. Obtain your key [here](https://victoriametrics.com/products/enterprise/trial/) for this tutorial or for enterprise use.
 - In the tutorial, we'll be using the following VictoriaMetrics components:
-  -  [VictoriaMetrics Single-Node](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) (v1.137.0)
-  -  [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/) (v1.137.0)
-  -  [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) (v1.137.0)
-- [Grafana](https://grafana.com/) (v.10.2.1)
+  -  [VictoriaMetrics Single-Node](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) (v1.145.0)
+  -  [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/) (v1.145.0)
+  -  [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) (v1.145.0)
+- [Grafana](https://grafana.com/) (v12.2.0)
 - [Docker](https://docs.docker.com/get-docker/) and [Docker Compose](https://docs.docker.com/compose/)
-- [Node exporter](https://github.com/prometheus/node_exporter#node-exporter) (v1.7.0) and [Alertmanager](https://prometheus.io/docs/alerting/latest/alertmanager/) (v0.27.0)
+- [Node exporter](https://github.com/prometheus/node_exporter#node-exporter) (v1.9.1) and [Alertmanager](https://prometheus.io/docs/alerting/latest/alertmanager/) (v0.28.1)
 
 ![typical setup diagram](guide-vmanomaly-vmalert_overview.webp)
 
@@ -323,7 +323,7 @@ Let's wrap it all up together into the `docker-compose.yml` file.
 services:
   vmagent:
     container_name: vmagent
-    image: victoriametrics/vmagent:v1.137.0
+    image: victoriametrics/vmagent:v1.145.0
     depends_on:
       - "victoriametrics"
     ports:
@@ -340,7 +340,7 @@ services:
 
   victoriametrics:
     container_name: victoriametrics
-    image: victoriametrics/victoria-metrics:v1.137.0
+    image: victoriametrics/victoria-metrics:v1.145.0
     ports:
       - 8428:8428
     volumes:
@@ -356,7 +356,7 @@ services:
 
   grafana:
     container_name: grafana
-    image: grafana/grafana-oss:10.2.1
+    image: grafana/grafana:12.2.0
     depends_on:
       - "victoriametrics"
     ports:
@@ -373,7 +373,7 @@ services:
 
   vmalert:
     container_name: vmalert
-    image: victoriametrics/vmalert:v1.137.0
+    image: victoriametrics/vmalert:v1.145.0
     depends_on:
       - "victoriametrics"
     ports:
@@ -395,7 +395,7 @@ services:
     restart: always
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.29.4
+    image: victoriametrics/vmanomaly:v1.29.5
     depends_on:
       - "victoriametrics"
     ports:
@@ -412,7 +412,7 @@ services:
       - "--licenseFile=/license"
   alertmanager:
     container_name: alertmanager
-    image: prom/alertmanager:v0.27.0
+    image: prom/alertmanager:v0.28.1
     volumes:
       - ./alertmanager.yml:/config/alertmanager.yml
     command:
@@ -424,7 +424,7 @@ services:
     restart: always
 
   node-exporter:
-    image: quay.io/prometheus/node-exporter:v1.7.0
+    image: quay.io/prometheus/node-exporter:v1.9.1
     container_name: node-exporter
     ports:
       - 9100:9100

docs/guides/guide-vmcluster-multiple-retention-setup/README.md

@@ -6,45 +6,348 @@ build:
 sitemap:
   disable: true
 ---
-**Objective**
 
-Setup Victoria Metrics Cluster with support of multiple retention periods within one installation.
+> [VictoriaMetrics Enterprise](https://docs.victoriametrics.com/victoriametrics/enterprise/) supports specifying multiple retentions for distinct sets of time series and tenants. If you are an Enterprise user, [configure multiple retentions directly through retention filters](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#retention-filters) instead of following this guide.
 
-**Enterprise Solution**
+This guide explains how to set up multiple retentions using an [open-source VictoriaMetrics Cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/).
 
-[VictoriaMetrics Enterprise](https://docs.victoriametrics.com/victoriametrics/enterprise/) supports specifying multiple retentions
-for distinct sets of time series and [tenants](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#multitenancy)
-via [retention filters](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#retention-filters).
+## Overview
 
-**Open Source Solution**
+VictoriaMetrics retains metrics by default for **1 month**. You can change data retention with the [`-retentionPeriod` command-line flag](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention), but this value applies to **all time series stored** on a given `vmstorage` node and cannot be customized per tenant or per metric in the open source version. 
 
-Community version of VictoriaMetrics supports only one retention period per `vmstorage` node via [-retentionPeriod](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention) command-line flag.
+The core idea of this guide is to run **separate logic groups of storages** (or even clusters) with individual `-retentionPeriod` settings, while still providing a single unified write and read path via vmagent and vmselect.
 
-A multi-retention setup can be implemented by dividing a [victoriametrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) into logical groups with different retentions.
+## Multi-Retention Architecture
 
-Example:
-Setup should handle 3 different retention groups 3months, 1year and 3 years.
-Solution contains 3 groups of vmstorages + vminserts and one group of vmselects. Routing is done by [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/)
-by [splitting data streams](https://docs.victoriametrics.com/victoriametrics/vmagent/#splitting-data-streams-among-multiple-systems). 
-The [-retentionPeriod](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention) sets how long to keep the metrics.
+To support multiple retentions with the open source version of VictoriaMetrics cluster, you can split the cluster into several logical groups of storage nodes. Each group is configured with a different `-retentionPeriod` and receives only the data that must follow that retention. 
 
-The diagram below shows a proposed solution
+Each storage group is connected to a separate vminsert, while a shared vmselect layer queries across all storage groups so that dashboards and alerts continue to see a single unified VictoriaMetrics backend.
 
 ![Setup](setup.webp)
 
-**Implementation Details**
+In the example used throughout this guide, the cluster is divided into three groups: 
 
-1. Groups of vminserts A know about only vmstorages A and this is explicitly specified via `-storageNode` [configuration](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#cluster-setup). 
-1. Groups of vminserts B know about only vmstorages B and this is explicitly specified via `-storageNode` [configuration](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#cluster-setup). 
-1. Groups of vminserts C know about only vmstorages C and this is explicitly specified via `-storageNode` [configuration](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#cluster-setup). 
-1. vmselect reads data from all vmstorage nodes via `-storageNode` [configuration](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#cluster-setup) 
-   with [deduplication](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#deduplication) setting equal to vmagent's scrape interval or minimum interval between collected samples. 
-1. vmagent routes incoming metrics to the given set of `vminsert` nodes using relabeling rules specified at `-remoteWrite.urlRelabelConfig` [configuration](https://docs.victoriametrics.com/victoriametrics/relabeling/).
+- Group A: 3-month retention.
+- Group B: 1-year retention.
+- Group C: 3-year retention. 
 
-**Multi-Tenant Setup**
+Metrics are routed to the appropriate vminsert group by splitting data streams in vmagent, so each time series is sent to exactly one retention group instead of being replicated to all groups. See [Deploying vmagent](https://docs.victoriametrics.com/guides/guide-vmcluster-multiple-retention-setup/#step3) for an example of label‑based routing that implements this split. An optional [vmauth](https://docs.victoriametrics.com/guides/guide-vmcluster-multiple-retention-setup/#additional-enhancements) layer can be added on top to restrict access to specific sub‑clusters or tenants while still keeping a unified write and read path.
 
-Every group of vmstorages can handle one tenant or multiple one. Different groups can have overlapping tenants. As vmselect reads from all vmstorage nodes, the data is aggregated on its level.
+## Implementing Multi-Retention on Kubernetes
 
-**Additional Enhancements**
+In this section, we'll install and configure the components for a multi-retention deployment of the VictoriaMetrics cluster. See [Kubernetes monitoring with VictoriaMetrics Cluster](https://docs.victoriametrics.com/guides/k8s-monitoring-via-vm-cluster/) for details on running VictoriaMetrics in Kubernetes.
 
-You can set up [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/) for routing data to the given vminsert group depending on the needed retention.
+Run the following command to add the VictoriaMetrics Helm repository:
+
+```shell
+helm repo add vm https://victoriametrics.github.io/helm-charts/
+helm repo update
+```
+
+### Step 1: Deploying storage groups {#step1}
+
+We'll create three storage groups. Each has a different retention period and disk size. Read [Understand Your Setup Size](https://docs.victoriametrics.com/guides/understand-your-setup-size/) to estimate how much space you will need for each group. The following table is shown as an example:
+
+
+| Group        | Retention Period | Total disk size       |
+|--------------|------------------|-----------------------|
+| `vmcluster-a`  | 3 months (`3M`)    | 80 Gi                 |
+| `vmcluster-b`  | 1 year (`1Y`)      | 300 Gi                |
+| `vmcluster-c`  | 3 years (`3Y`)     | 900 Gi                |
+
+Create a Helm values file for Group A.
+
+```shell
+cat <<EOF > vmcluster-a.yaml
+vmstorage:
+  enabled: true
+  replicaCount: 1
+  persistence:
+    size: 80Gi
+  extraArgs:
+    retentionPeriod: 3M
+  podLabels:
+    retention-group: a
+
+vminsert:
+  enabled: true
+  podLabels:
+    retention-group: a
+
+vmselect:
+  enabled: false
+EOF
+```
+
+The values file above creates vminsert and vmstorage services while turning off vmselect, which we'll deploy separately. The `retentionPeriod` flag configures how long data is kept in this group.
+
+Create the values files for Group B and Group C:
+
+```shell
+cat <<EOF > vmcluster-b.yaml
+vmstorage:
+  enabled: true
+  replicaCount: 1
+  persistence:
+    size: 300Gi
+  extraArgs:
+    retentionPeriod: 1y
+  podLabels:
+    retention-group: b
+
+vminsert:
+  enabled: true
+  podLabels:
+    retention-group: b
+
+vmselect:
+  enabled: false
+EOF
+
+
+cat <<EOF > vmcluster-c.yaml
+vmstorage:
+  enabled: true
+  replicaCount: 1
+  persistence:
+    size: 900Gi
+  extraArgs:
+    retentionPeriod: 3y
+  podLabels:
+    retention-group: c
+
+vminsert:
+  enabled: true
+  podLabels:
+    retention-group: c
+
+vmselect:
+  enabled: false
+EOF
+```
+
+Deploy the three storage groups with:
+
+```shell
+helm upgrade --install vmcluster-a vm/victoria-metrics-cluster -f vmcluster-a.yaml
+helm upgrade --install vmcluster-b vm/victoria-metrics-cluster -f vmcluster-b.yaml
+helm upgrade --install vmcluster-c vm/victoria-metrics-cluster -f vmcluster-c.yaml
+
+# Wait for all storage pods to be ready
+kubectl rollout status statefulset -l app.kubernetes.io/instance=vmcluster-a
+kubectl rollout status statefulset -l app.kubernetes.io/instance=vmcluster-b
+kubectl rollout status statefulset -l app.kubernetes.io/instance=vmcluster-c
+```
+
+### Step 2: Deploying vmselect {#step2}
+
+Next, we'll deploy a vmselect service to route queries to the storage groups.
+
+Create a Helm values file with:
+
+```shell
+cat <<EOF >vmselect.yaml
+vmstorage:
+  enabled: false
+
+vminsert:
+  enabled: false
+
+vmselect:
+  enabled: true
+  replicaCount: 1
+  suppressStorageFQDNsRender: true
+  extraArgs:
+    # Each list item is a single -storageNode flag. In this example, there is
+    # one vmstorage pod per retention group, so each entry contains a single host.
+    # If you run multiple pods per group, list them as comma-separated hosts
+    # in the same -storageNode value.
+    #
+    # The FQDN format is:
+    #   <pod>.<svc>.default.svc
+    # where pod = <release>-victoria-metrics-cluster-vmstorage-<N>
+    # and   svc = <release>-victoria-metrics-cluster-vmstorage
+    storageNode:
+      - "vmcluster-a-victoria-metrics-cluster-vmstorage-0.vmcluster-a-victoria-metrics-cluster-vmstorage.default.svc:8401"
+      - "vmcluster-b-victoria-metrics-cluster-vmstorage-0.vmcluster-b-victoria-metrics-cluster-vmstorage.default.svc:8401"
+      - "vmcluster-c-victoria-metrics-cluster-vmstorage-0.vmcluster-c-victoria-metrics-cluster-vmstorage.default.svc:8401"
+EOF
+```
+
+Let's break down the file above:
+
+- Deploys vmselect as a separate Helm release. 
+- Disables vminsert and vmstorage as these services were already deployed in Step 1.
+- `suppressStorageFQDNsRender: true` turns off automatic FQDN generation for storage nodes. By default, the Helm chart auto-generates `-storageNodes` flags, but since `vmstorage` has been disabled, we need to supply them manually in `extraArgs`.
+- In `extraArgs.storageNode:` we define the vmstorage endpoints for queries. On querying, vmselect merges results across all the specified vmstorages to provide a unified view of the data.
+
+Deploy the `vmselect` release with:
+
+```shell
+helm upgrade --install vmselect vm/victoria-metrics-cluster -f vmselect.yaml
+```
+
+### Step 3: Deploying vmagent {#step3}
+
+We'll use `vmagent` to route incoming metrics to the correct retention group. For example, we can use a `retention` label for mapping metrics to storage groups in the following way:
+
+| `retention` label | Storage Group |
+|-------------------|--------------|
+| `"3mo"`           | `vmcluster-a` |
+| `"1yr"`           | `vmcluster-b` |
+| `"3yr"`           | `vmcluster-c` |
+
+
+Create the values file for vmagent:
+
+```shell
+cat <<EOF >vmagent.yaml
+service:
+  enabled: true
+remoteWrite:
+  # Group A: receives metrics with retention="3mo"
+  - url: http://vmcluster-a-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write
+    urlRelabelConfig:
+      - if: '{retention="3mo"}'
+        action: keep
+  # Group B: receives metrics with retention="1yr"
+  - url: http://vmcluster-b-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write
+    urlRelabelConfig:
+      - if: '{retention="1yr"}'
+        action: keep
+  # Group C: receives metrics with retention="3yr"
+  - url: http://vmcluster-c-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write
+    urlRelabelConfig:
+      - if: '{retention="3yr"}'
+        action: keep
+EOF
+```
+
+> Metrics without a matching `retention` label are silently dropped by the `keep` rules. You must ensure that every metric is labeled, or use a different routing configuration.
+
+Now deploy the vmagent release:
+
+```shell
+helm upgrade --install vmagent vm/victoria-metrics-agent -f vmagent.yaml
+```
+
+Wait for vmagent to become ready:
+
+```shell
+kubectl rollout status deploy/vmagent-victoria-metrics-agent
+```
+
+### Step 4: Verification
+
+We can send test data to verify that the data is flowing to the correct storage group.
+
+First, port-forward vmagent and vmselect:
+
+```shell
+VMAGENT_SVC=$(kubectl get svc -l app.kubernetes.io/instance=vmagent -o jsonpath='{.items[0].metadata.name}')
+kubectl port-forward "svc/$VMAGENT_SVC" 8429 &
+
+VMSELECT_SVC=$(kubectl get svc -l app.kubernetes.io/instance=vmselect -o jsonpath='{.items[0].metadata.name}')
+kubectl port-forward "svc/$VMSELECT_SVC" 8481 &
+```
+
+Send test metrics directly to vmagent's HTTP endpoint to exercise all three retention labels:
+
+```shell
+POD=$(kubectl get pod -l app.kubernetes.io/instance=vmagent -o jsonpath='{.items[0].metadata.name}')
+
+for retention in 3mo 1yr 3yr; do
+  kubectl exec "$POD" -- wget -qO- --post-data="test_routing{retention=\"${retention}\"} 1.0" \
+    "http://127.0.0.1:8429/api/v1/import/prometheus"
+done
+```
+
+Query the data back from vmselect (it may take around 30-60 seconds for new data to be available for queries):
+
+```shell
+for retention in 3mo 1yr 3yr; do
+  echo "-> retention=${retention}"
+  curl -s "http://localhost:8481/select/0/prometheus/api/v1/query" \
+    --data-urlencode "query=test_routing{retention=\"${retention}\"}"
+  echo
+done
+```
+
+You can also check that vmagent is forwarding data to all three groups:
+
+```shell
+curl -s http://localhost:8429/metrics | grep vmagent_remotewrite_blocks_sent_total
+```
+
+Each `url="N:secret-url"` corresponds to one `remoteWrite` entry (N=1 for Group A, N=2 for Group B, N=3 for Group C). Non-zero values confirm data is flowing.
+
+## Alternative Routing by Existing Labels
+
+The example setup above relies on a synthetic `retention` label to exist in every incoming metric.
+
+If having a `retention` label in every metric isn't practical, you can, as an alternative, rely on existing labels to map data to the correct storage group.
+
+The following example configures vmagent to route metrics based on the `environment` and `team` labels:
+
+```yaml
+# vmagent.yaml
+remoteWrite:
+  # send dev and staging data to Group A
+  - url: "http://vmcluster-a-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write"
+    urlRelabelConfig:
+      - if: {environment=~"dev|staging"}
+        action: keep
+  # send prod data to Group B
+  - url: "http://vmcluster-b-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write"
+    urlRelabelConfig:
+      - if: {environment=~"prod|production"}
+        action: keep
+  # send data from Infra and SRE teams to Group C
+  - url: "http://vmcluster-c-victoria-metrics-cluster-vminsert:8480/insert/0/prometheus/api/v1/write"
+    urlRelabelConfig:
+      - if: {team=~"infra|sre"}
+        action: keep
+```
+
+> Metrics that do not match any of the `keep` rules are dropped in the configuration above.
+
+## Additional Enhancements
+
+You can set up [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/) to route data to the specified vminsert group based on the required retention or to restrict which data different users can query.
+
+The following [`-auth.config`](https://docs.victoriametrics.com/victoriametrics/vmauth/#quick-start) example exposes the same vmselect backend via vmauth with two users using basic auth:
+
+- `admin`: can query **all** data across all retention groups.
+- `dev`: can query **only** time series that have `team="dev"` label, enforced via the `extra_label` query argument.
+
+```yaml
+users:
+  # User with access to all data across all retention groups
+  - username: "admin"
+    password: "foo"
+    url_map:
+      - src_paths:
+          - "/api/v1/query"
+          - "/api/v1/query_range"
+          - "/api/v1/series"
+          - "/api/v1/labels"
+          - "/api/v1/label/.+/values"
+        # vmselect service that aggregates all vmstorage groups
+        url_prefix: "http://vmselect-victoria-metrics-cluster-vmselect:8481/select/0/prometheus"
+
+  # User restricted to Dev team data only
+  - username: "dev"
+    password: "bar"
+    url_map:
+      - src_paths:
+          - "/api/v1/query"
+          - "/api/v1/query_range"
+          - "/api/v1/series"
+          - "/api/v1/labels"
+          - "/api/v1/label/.+/values"
+        # Same vmselect backend, but enforce label filter at query time
+        # by adding extra_label=team=dev to every proxied request
+        url_prefix: "http://vmselect-victoria-metrics-cluster-vmselect:8481/select/0/prometheus/?extra_label=team=dev"
+```
+
+This is useful for restricting access by team, environment, or tenant without changing the underlying storage topology.

docs/victoriametrics/CONTRIBUTING.md

@@ -28,7 +28,7 @@ If you like VictoriaMetrics and want to contribute, then it would be great:
 ## Issues
 
 When making a new issue, make sure to create no duplicates. Use GitHub search to find whether similar issues exist already.
-The new issue should be written in English and contain a concise description of the problem and the environment where it exists.
+The new issue should be written in English and contain concise description of the problem and environment where it exists.
 We'd very much prefer to have a specific use-case included in the description, since it could have workaround or alternative solutions.
 
 When looking for an issue to contribute, always prefer working on [bugs](https://github.com/VictoriaMetrics/VictoriaMetrics/issues?q=is%3Aopen+is%3Aissue+label%3Abug)
@@ -48,7 +48,7 @@ We use [labels](https://docs.github.com/en/issues/using-labels-and-milestones-to
 1. `need more info`, assigned to issues that require elaboration from the issue creator.
   For example, if we weren't able to reproduce the reported bug based on the ticket description then we ask additional
   questions which could help to reproduce the issue and add `need more info` label. This label helps other maintainers
-  to understand that this issue wasn't forgotten but waits for the feedback from the user.
+  to understand that this issue wasn't forgotten but waits for the feedback from user.
 1. `completed`, assigned to issues that required code changes and those changes were merged to upstream, but not released yet.
   Once a release is made, maintainers go through all labeled issues, leave a comment about the new release, and close the issue.
 1. `vmui`, assigned to issues related to [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui) or [VictoriaLogs webui](https://docs.victoriametrics.com/victorialogs/querying/#web-ui)
@@ -63,31 +63,32 @@ Pull requests requirements:
 1. Don't use `master` branch for making PRs, as it makes it impossible for reviewers to modify the changes.
 1. All commits need to be [signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits).
 1. Pull request title should be prefixed with `<dir>/<component>:` to show what component has been changed, i.e. `app/vmalert: fix...`.
-   Pull request description should contain a clear and concise description of what was done, why it is needed and for what purpose.
+   Pull request description should contain clear and concise description of what was done, why it is needed and for what purpose.
    Use clear language, so reviewers can quickly understand the change and its impact.
 1. A link to the issue(s) related to the change, if any. Use `Fixes [issue link]` if the PR resolves the issue, or `Related to [issue link]` for reference.
 1. Tests proving that the change is effective. Tests are expected for non-trivial new functionality or non-trivial modifications.
    Bug fixes must include tests unless a maintainer explicitly agrees otherwise.
-   See [this style guide](https://itnext.io/f-tests-as-a-replacement-for-table-driven-tests-in-go-8814a8b19e9e) for tests. See [this section](#testing) for how to run tests.
+   See [this style guide](https://itnext.io/f-tests-as-a-replacement-for-table-driven-tests-in-go-8814a8b19e9e) for tests.
+   To run tests and code checks locally, execute commands `make test-full` and `make check-all`.
 1. Try to not extend the scope of the pull requests outside the issue, do not make unrelated changes.
-1. Update [docs](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/docs) if needed. For example, adding a new flag or changing the behavior of existing flags or features
+1. Update [docs](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/docs) if needed. For example, adding a new flag or changing behavior of existing flags or features
    requires reflecting these changes in the documentation. For new features add `{{%/* available_from "#" */%}}` shortcode to the documentation.
    It will be later automatically replaced with an actual release version.
 1. A line in the [changelog](https://docs.victoriametrics.com/victoriametrics/changelog/#tip) mentioning the change and related issue in a way
    that would be clear to other readers even if they don't have the full context.
-1. Avoid modifying code in the `/vendor` folder manually, even when the vendored package originates from the VictoriaMetrics GitHub organization.
+1. Avoid modifying code in the `/vendor` folder manually, even when the vendored package originates are from the VictoriaMetrics GitHub organization.
    For instance, VictoriaLogs vendors packages under the `/lib` folder from VictoriaMetrics, and VictoriaTraces vendors the `/lib/logstorage` package from VictoriaLogs.
-   Submit a pull request to the upstream repository first. Afterward, a separate pull request can be opened to update the version of the vendored folder in the downstream repository.
+   Submit a pull request to the upstream repository first. Afterward, a separate pull request can be opened to update the version of the vendored folder in downstream repository.
    * For common packages, the vendored package can be updated with this command: `go get <dependency>@vX.Y.Z`.  
    * For VictoriaMetrics packages, use `go get <dependency>@canonical_commit_hash`. 
    Finally, run `go mod tidy` and `go mod vendor` to update `go.mod`, `go.sum`, and `/vendor`.
 1. Ping reviewers who you think have the best expertise on the matter.
 
-See a good example of a [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6487).
+See good example of a [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6487).
 
 ## Merging Pull Request
 
-The person who merges the Pull Request is responsible for satisfying the requirements below:
+The person who merges the Pull Request is responsible for satisfying requirements below:
 
 1. Make sure that PR satisfies [Pull Request checklist](https://docs.victoriametrics.com/victoriametrics/contributing/#pull-request-checklist),
    it is approved by at least one reviewer, all CI checks are green.
@@ -96,9 +97,9 @@ The person who merges the Pull Request is responsible for satisfying the require
 1. If applicable, cherry-pick the change to [LTS release lines](https://docs.victoriametrics.com/victoriametrics/lts-releases/)
    and mention in the PR comment what was or wasn't cherry-picked.
 1. Update related issues with a meaningful message of what has changed and when it will be
-   released. _This helps users to understand the change without reading the PR._
+   released. _This helps users to understand the change without reading PR._
 1. Add label `completed` to related issues.
-1. Do not close related tickets until the release is made. If the ticket was auto-closed by GitHub or a user - re-open it.
+1. Do not close related tickets until release is made. If ticket was auto-closed by GitHub or user - re-open it.
 
 ## KISS principle
 
@@ -114,9 +115,9 @@ We are open to third-party pull requests provided they follow [KISS design princ
 - Minimize the number of moving parts in the distributed system.
 - Avoid automated decisions, which may hurt cluster availability, consistency, performance or debuggability.
 
-Adhering to the `KISS` principle, simplifies the resulting code and architecture so it can be reviewed, understood and debugged by a wider audience.
+Adhering to `KISS` principle, simplifies the resulting code and architecture so it can be reviewed, understood and debugged by a wider audience.
 
-Due to `KISS`, [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) has none of the following "features" popular in distributed computing:
+Due to `KISS`, [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) has none of the following "features" popular in distributed computing world:
 
 - Fragile gossip protocols. See [failed attempt in Thanos](https://github.com/improbable-eng/thanos/blob/030bc345c12c446962225221795f4973848caab5/docs/proposals/completed/201809_gossip-removal.md).
 - Hard-to-understand-and-implement-properly [Paxos protocols](https://www.quora.com/In-distributed-systems-what-is-a-simple-explanation-of-the-Paxos-algorithm).
@@ -125,17 +126,3 @@ Due to `KISS`, [cluster version of VictoriaMetrics](https://docs.victoriametrics
 - Automatic cluster resizing, which may cost you a lot of money if improperly configured.
 - Automatic discovering and addition of new nodes in the cluster, which may mix data between dev and prod clusters :)
 - Automatic leader election, which may result in split brain disaster on network errors.
-
-## Testing
-
-We recommend running the following sequence of checks and tests before submitting a pull request:
-```sh
-# run static checks
-make check-all
-
-# run unit test
-make test-full
-
-# run integration tests
-make apptest
-```

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,6 +26,11 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): log calls to [/api/v1/admin/tsdb/delete_series](https://docs.victoriametrics.com/victoriametrics/url-examples/#apiv1admintsdbdelete_series) API handler. This should help to identify events of metrics deletion from the database. 
+
+
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix issue with producing aggregated samples with identical timestamps between flushes. See PR [#10808](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10808) for details.
+
 ## [v1.145.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.145.0)
 
 Released at 2026-06-08

lib/flagutil/password.go

@@ -121,7 +121,7 @@ func (p *Password) initRandomValue() {
 	_, err := io.ReadFull(rand.Reader, buf[:])
 	if err != nil {
 		// cannot use lib/logger here, since it can be uninitialized yet
-		panic(fmt.Errorf("FATAL: cannot read random data: %s", err))
+		panic(fmt.Errorf("FATAL: cannot read random data: %w", err))
 	}
 	s := string(buf[:])
 	p.value.Store(&s)

lib/jwt/key.go

@@ -16,7 +16,7 @@ func ParseKey(key []byte) (any, error) {
 
 	k, err := x509.ParsePKIXPublicKey(b.Bytes)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse key %q: %v", key, err)
+		return nil, fmt.Errorf("failed to parse key %q: %w", key, err)
 	}
 
 	return k, nil

lib/prompb/prompb_timing_test.go

@@ -14,7 +14,7 @@ func BenchmarkWriteRequestUnmarshalProtobuf(b *testing.B) {
 		wru := &WriteRequestUnmarshaler{}
 		for pb.Next() {
 			if _, err := wru.UnmarshalProtobuf(data); err != nil {
-				panic(fmt.Errorf("unexpected error: %s", err))
+				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}
 	})

lib/promscrape/discovery/ovhcloud/dedicated_server.go

@@ -97,12 +97,12 @@ func getDedicatedServerDetails(cfg *apiConfig, dedicatedServerName string) (*ded
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	var dedicatedServerDetails dedicatedServer
 	if err = json.Unmarshal(resp, &dedicatedServerDetails); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	// get IPs for this dedicated server.
@@ -113,12 +113,12 @@ func getDedicatedServerDetails(cfg *apiConfig, dedicatedServerName string) (*ded
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	var ips []string
 	if err = json.Unmarshal(resp, &ips); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	// handle different IP formats
@@ -141,11 +141,11 @@ func getDedicatedServerList(cfg *apiConfig) ([]string, error) {
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	if err = json.Unmarshal(resp, &dedicatedServerList); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	return dedicatedServerList, nil

lib/promscrape/discovery/ovhcloud/vps.go

@@ -117,12 +117,12 @@ func getVPSDetails(cfg *apiConfig, vpsName string) (*virtualPrivateServer, error
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	var vpsDetails virtualPrivateServer
 	if err = json.Unmarshal(resp, &vpsDetails); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	// get IPs for this vps.
@@ -133,12 +133,12 @@ func getVPSDetails(cfg *apiConfig, vpsName string) (*virtualPrivateServer, error
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	var ips []string
 	if err = json.Unmarshal(resp, &ips); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	// handle different IP formats
@@ -162,12 +162,12 @@ func getVPSList(cfg *apiConfig) ([]string, error) {
 		request.Header, _ = getAuthHeaders(cfg, request.Header, cfg.client.APIServer(), reqPath)
 	})
 	if err != nil {
-		return nil, fmt.Errorf("request %s error: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot process %s: %w", reqPath, err)
 	}
 
 	var vpsList []string
 	if err = json.Unmarshal(resp, &vpsList); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal %s response: %v", reqPath, err)
+		return nil, fmt.Errorf("cannot unmarshal response from %s: %w", reqPath, err)
 	}
 
 	return vpsList, nil

lib/promscrape/discovery/puppetdb/api.go

@@ -33,7 +33,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 	}
 	parsedURL, err := url.Parse(sdc.URL)
 	if err != nil {
-		return nil, fmt.Errorf("parse URL %s error: %v", sdc.URL, err)
+		return nil, fmt.Errorf("cannot parse %s: %w", sdc.URL, err)
 	}
 	if parsedURL.Scheme != "http" && parsedURL.Scheme != "https" {
 		return nil, fmt.Errorf("URL %s scheme must be 'http' or 'https'", sdc.URL)

lib/promscrape/discovery/yandexcloud/api.go

@@ -221,7 +221,7 @@ func getIAMToken(cfg *apiConfig) (*iamToken, error) {
 	body := bytes.NewBuffer(passport)
 	resp, err := cfg.client.Post(iamURL, "application/json", body)
 	if err != nil {
-		return nil, fmt.Errorf("cannot send request to yandex cloud iam api %q: %s", iamURL, err)
+		return nil, fmt.Errorf("cannot send request to yandex cloud iam api %q: %w", iamURL, err)
 	}
 	data, err := readResponseBody(resp, iamURL)
 	if err != nil {

lib/promscrape/scrapework_test.go

@@ -667,11 +667,11 @@ func TestScrapeWorkScrapeInternalStreamConcurrency(t *testing.T) {
 	}
 
 	generateScrape := func(n int) string {
-		w := strings.Builder{}
+		var w strings.Builder
 		for i := range n {
-			w.WriteString(fmt.Sprintf("fooooo_%d 1\n", i))
+			fmt.Fprintf(&w, "fooooo_%d 1\n", i)
 			if i%100 == 0 {
-				w.WriteString(fmt.Sprintf("# HELP fooooo_%d This is a test\n", i))
+				fmt.Fprintf(&w, "# HELP fooooo_%d This is a test\n", i)
 			}
 		}
 		return w.String()
@@ -1005,9 +1005,9 @@ func TestSendStaleSeries(t *testing.T) {
 		}
 	}
 	generateScrape := func(n int) string {
-		w := strings.Builder{}
+		var w strings.Builder
 		for i := range n {
-			w.WriteString(fmt.Sprintf("foo_%d 1\n", i))
+			fmt.Fprintf(&w, "foo_%d 1\n", i)
 		}
 		return w.String()
 	}

lib/protoparser/datadogv2/parser.go

@@ -186,7 +186,7 @@ func (s *Series) unmarshalProtobuf(src []byte) (err error) {
 			}
 			pt := &points[len(points)-1]
 			if err := pt.unmarshalProtobuf(data); err != nil {
-				return fmt.Errorf("cannot unmarshal point: %s", err)
+				return fmt.Errorf("cannot unmarshal point: %w", err)
 			}
 		case 1:
 			data, ok := fc.MessageData()

lib/protoparser/opentelemetry/firehose/parser.go

@@ -30,7 +30,7 @@ func ProcessRequestBody(b []byte) ([]byte, error) {
 		}
 	}
 	if err := json.Unmarshal(b, &req); err != nil {
-		return nil, fmt.Errorf("cannot unmarshal Firehose JSON in request body: %s", err)
+		return nil, fmt.Errorf("cannot unmarshal Firehose JSON in request body: %w", err)
 	}
 
 	var dst []byte

lib/protoparser/zabbixconnector/parser.go

@@ -99,17 +99,17 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
 
 	n, err := getFloat64(o, "value")
 	if err != nil {
-		return fmt.Errorf("missing `value` element, %s", err)
+		return fmt.Errorf("missing `value` element: %w", err)
 	}
 	r.Value = n
 
 	cl, err := getInt64(o, "clock")
 	if err != nil {
-		return fmt.Errorf("missing `clock` element, %s", err)
+		return fmt.Errorf("missing `clock` element: %w", err)
 	}
 	ns, err := getInt64(o, "ns")
 	if err != nil {
-		return fmt.Errorf("missing `ns` element, %s", err)
+		return fmt.Errorf("missing `ns` element: %w", err)
 	}
 	// clock - Number of seconds since Epoch to the moment when value was collected (integer part).
 	// ns - Number of nanoseconds to be added to clock to get a precise value collection time.
@@ -121,7 +121,7 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
 	if len(groupValue) != 0 {
 		groups, err := getArray(o, "groups")
 		if err != nil {
-			return fmt.Errorf("missing `groups` element, %s", err)
+			return fmt.Errorf("missing `groups` element: %w", err)
 		}
 		for _, g := range groups {
 			k := g.GetStringBytes()
@@ -141,7 +141,7 @@ func (r *Row) unmarshal(o *fastjson.Value) error {
 
 	itemTags, err := getArray(o, "item_tags")
 	if err != nil {
-		return fmt.Errorf("missing `item_tags` element, %s", err)
+		return fmt.Errorf("missing `item_tags` element: %w", err)
 	}
 
 	if len(duplicateTagsSeparator) == 0 { // Do not merge tags

lib/snapshot/snapshot.go

@@ -71,9 +71,9 @@ func Create(ctx context.Context, createSnapshotURL string) (string, error) {
 		return snap.Snapshot, nil
 	}
 	if snap.Status == "error" {
-		return "", errors.New(snap.Msg)
+		return "", fmt.Errorf("snapshot status: %q; msg: %q", snap.Status, snap.Msg)
 	}
-	return "", fmt.Errorf("unknown status: %v", snap.Status)
+	return "", fmt.Errorf("snapshot status unknown: %q", snap.Status)
 }
 
 // Delete deletes a snapshot via the provided api endpoint
@@ -121,14 +121,14 @@ func Delete(ctx context.Context, deleteSnapshotURL string, snapshotName string)
 	if snap.Status == "error" {
 		return errors.New(snap.Msg)
 	}
-	return fmt.Errorf("unknown status: %v", snap.Status)
+	return fmt.Errorf("snapshot status unknown: %q", snap.Status)
 }
 
 // GetHTTPClient returns a new HTTP client configured for snapshot operations.
 func GetHTTPClient() (*http.Client, error) {
 	tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vm_snapshot_client")
 	if err != nil {
-		return nil, fmt.Errorf("failed to create transport: %s", err)
+		return nil, fmt.Errorf("failed to create transport: %w", err)
 	}
 	hc := &http.Client{
 		Transport: tr,

lib/storage/index_db.go

@@ -649,7 +649,7 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
 		} else {
 			_, key, err := unmarshalCompositeTagKey(labelName)
 			if err != nil {
-				return nil, fmt.Errorf("cannot unmarshal composite tag key: %s", err)
+				return nil, fmt.Errorf("cannot unmarshal composite tag key: %w", err)
 			}
 			lns[string(key)] = struct{}{}
 		}

lib/storage/search.go

@@ -468,15 +468,21 @@ func (tf *TagFilter) Unmarshal(src []byte) ([]byte, error) {
 	return src, nil
 }
 
-// String returns string representation of the search query.
+// String returns string representation of the search query: tag filters and time range.
 func (sq *SearchQuery) String() string {
+	start := TimestampToHumanReadableFormat(sq.MinTimestamp)
+	end := TimestampToHumanReadableFormat(sq.MaxTimestamp)
+	a := sq.FiltersString()
+	return fmt.Sprintf("filters=%s, timeRange=[%s..%s]", a, start, end)
+}
+
+// FiltersString returns string representation of the tag filters.
+func (sq *SearchQuery) FiltersString() []string {
 	a := make([]string, len(sq.TagFilterss))
 	for i, tfs := range sq.TagFilterss {
 		a[i] = tagFiltersToString(tfs)
 	}
-	start := TimestampToHumanReadableFormat(sq.MinTimestamp)
-	end := TimestampToHumanReadableFormat(sq.MaxTimestamp)
-	return fmt.Sprintf("filters=%s, timeRange=[%s..%s]", a, start, end)
+	return a
 }
 
 func tagFiltersToString(tfs []TagFilter) string {

lib/streamaggr/deduplicator.go

@@ -231,6 +231,7 @@ func (d *Deduplicator) flush(pushFunc PushFunc) {
 		logger.Warnf("deduplication couldn't be finished in the configured dedupInterval=%s; it took %.03fs; "+
 			"possible solutions: increase dedupInterval; reduce samples' ingestion rate", d.interval, duration.Seconds())
 	}
+	deadlineTime = deadlineTime.Add(d.interval)
 	for time.Now().After(deadlineTime) {
 		deadlineTime = deadlineTime.Add(d.interval)
 	}

lib/streamaggr/histogram_bucket.go

@@ -14,16 +14,10 @@ func (av *histogramBucketAggrValue) pushSample(_ aggrConfig, sample *pushSample,
 	av.h.Update(sample.value)
 }
 
-func (av *histogramBucketAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, _ bool) {
-	ac := c.(*histogramBucketAggrConfig)
-	shared := av.shared
-	if ac.useSharedState {
-		shared.Merge(&av.h)
-		av.h.Reset()
-	} else {
-		shared = &av.h
-	}
-	shared.VisitNonZeroBuckets(func(vmrange string, count uint64) {
+func (av *histogramBucketAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string, _ bool) {
+	av.shared.Merge(&av.h)
+	av.h.Reset()
+	av.shared.VisitNonZeroBuckets(func(vmrange string, count uint64) {
 		ctx.appendSeriesWithExtraLabel(key, "histogram_bucket", float64(count), "vmrange", vmrange)
 	})
 }
@@ -32,26 +26,17 @@ func (av *histogramBucketAggrValue) state() any {
 	return av.shared
 }
 
-func newHistogramBucketAggrConfig(useSharedState bool) aggrConfig {
-	return &histogramBucketAggrConfig{
-		useSharedState: useSharedState,
-	}
+func newHistogramBucketAggrConfig() aggrConfig {
+	return &histogramBucketAggrConfig{}
 }
 
-type histogramBucketAggrConfig struct {
-	useSharedState bool
-}
+type histogramBucketAggrConfig struct{}
 
-func (ac *histogramBucketAggrConfig) getValue(s any) aggrValue {
-	var shared *metrics.Histogram
-	if ac.useSharedState {
-		if s == nil {
-			shared = &metrics.Histogram{}
-		} else {
-			shared = s.(*metrics.Histogram)
-		}
+func (*histogramBucketAggrConfig) getValue(s any) aggrValue {
+	if s == nil {
+		s = &metrics.Histogram{}
 	}
 	return &histogramBucketAggrValue{
-		shared: shared,
+		shared: s.(*metrics.Histogram),
 	}
 }

lib/streamaggr/increase.go

@@ -0,0 +1,109 @@
+package streamaggr
+
+import (
+	"fmt"
+
+	"github.com/VictoriaMetrics/metrics"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+type increaseLastValue struct {
+	value          float64
+	timestamp      int64
+	deleteDeadline int64
+}
+
+type increaseAggrConfig struct {
+	keepFirstSample bool
+
+	// The first sample per each new series is ignored until this unix timestamp deadline in seconds even if keepFirstSample is set.
+	// This allows avoiding an initial spike of the output values at startup when new time series
+	// cannot be distinguished from already existing series. This is tracked with ignoreFirstSampleDeadline.
+	ignoreFirstSampleDeadline uint64
+	counterResetsTotal        *metrics.Counter
+}
+
+type increaseAggrValue struct {
+	total  *float64
+	shared map[string]increaseLastValue
+}
+
+func (av *increaseAggrValue) pushSample(c aggrConfig, sample *pushSample, key string, deleteDeadline int64) {
+	ac := c.(*increaseAggrConfig)
+	currentTime := fasttime.UnixTimestamp()
+	keepFirstSample := ac.keepFirstSample && currentTime >= ac.ignoreFirstSampleDeadline
+	lv, ok := av.shared[key]
+	if av.total == nil {
+		av.total = new(float64)
+	}
+	if ok {
+		if sample.timestamp < lv.timestamp {
+			// Skip out of order sample
+			return
+		}
+		if sample.value >= lv.value {
+			*av.total += sample.value - lv.value
+		} else {
+			// counter reset
+			*av.total += sample.value
+			ac.counterResetsTotal.Inc()
+		}
+	} else if keepFirstSample {
+		*av.total += sample.value
+	}
+	lv.value = sample.value
+	lv.timestamp = sample.timestamp
+	lv.deleteDeadline = deleteDeadline
+	key = bytesutil.InternString(key)
+	av.shared[key] = lv
+}
+
+func (av *increaseAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, isLast bool) {
+	ac := c.(*increaseAggrConfig)
+	for lk, lv := range av.shared {
+		if ctx.flushTimestamp > lv.deleteDeadline || isLast {
+			delete(av.shared, lk)
+		}
+	}
+	if av.total == nil {
+		return
+	}
+	total := *av.total
+	av.total = nil
+	ctx.appendSeries(key, ac.getSuffix(), total)
+}
+
+func (av *increaseAggrValue) state() any {
+	return av.shared
+}
+
+func newIncreaseAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, keepFirstSample bool) aggrConfig {
+	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + ignoreFirstSampleIntervalSecs
+	cfg := &increaseAggrConfig{
+		keepFirstSample:           keepFirstSample,
+		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
+	}
+	cfg.counterResetsTotal = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_counter_resets_total{%s}`, metricLabels))
+	return cfg
+}
+
+func (*increaseAggrConfig) getValue(s any) aggrValue {
+	var shared map[string]increaseLastValue
+	if s == nil {
+		shared = make(map[string]increaseLastValue)
+	} else {
+		shared = s.(map[string]increaseLastValue)
+	}
+	return &increaseAggrValue{
+		shared: shared,
+	}
+}
+
+func (ac *increaseAggrConfig) getSuffix() string {
+	if ac.keepFirstSample {
+		return "increase"
+	}
+	return "increase_prometheus"
+}

lib/streamaggr/output.go

@@ -75,6 +75,9 @@ func (ao *aggrOutputs) pushSamples(samples []pushSample, deleteDeadline int64, i
 				outputs = av.blue
 			}
 			for idx, o := range outputs {
+				if o == nil {
+					o = av.blue[idx]
+				}
 				o.pushSample(ao.configs[idx], sample, inputKey, deleteDeadline)
 			}
 			av.deleteDeadline = deleteDeadline
@@ -112,6 +115,9 @@ func (ao *aggrOutputs) flushState(ctx *flushCtx) {
 			outputs = av.blue
 		}
 		for i, o := range outputs {
+			if o == nil {
+				o = av.blue[i]
+			}
 			o.flush(ao.configs[i], ctx, outputKey, ctx.isLast)
 		}
 		av.mu.Unlock()

lib/streamaggr/streamaggr.go

@@ -609,7 +609,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	outputsSeen := make(map[string]struct{}, len(cfg.Outputs))
 	for i, output := range cfg.Outputs {
 		outputMetricLabels := fmt.Sprintf(`output=%q,name=%q,path=%q,url=%q,position="%d"`, output, name, path, alias, aggrID)
-		ac, err := newOutputConfig(ms, outputMetricLabels, output, outputsSeen, useSharedState, ignoreFirstSampleInterval)
+		ac, err := newOutputConfig(ms, outputMetricLabels, output, outputsSeen, ignoreFirstSampleInterval)
 		if err != nil {
 			return nil, err
 		}
@@ -716,7 +716,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	return a, nil
 }
 
-func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen map[string]struct{}, useSharedState bool, ignoreFirstSampleInterval time.Duration) (aggrConfig, error) {
+func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen map[string]struct{}, ignoreFirstSampleInterval time.Duration) (aggrConfig, error) {
 	// check for duplicated output
 	if _, ok := outputsSeen[output]; ok {
 		return nil, fmt.Errorf("`outputs` list contains duplicate aggregation function: %s", output)
@@ -760,11 +760,11 @@ func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen m
 	case "count_series":
 		return newCountSeriesAggrConfig(), nil
 	case "histogram_bucket":
-		return newHistogramBucketAggrConfig(useSharedState), nil
+		return newHistogramBucketAggrConfig(), nil
 	case "increase":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true, true), nil
+		return newIncreaseAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true), nil
 	case "increase_prometheus":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true, false), nil
+		return newIncreaseAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false), nil
 	case "last":
 		return newLastAggrConfig(), nil
 	case "max":
@@ -782,9 +782,9 @@ func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen m
 	case "sum_samples":
 		return newSumSamplesAggrConfig(), nil
 	case "total":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false, true), nil
+		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true), nil
 	case "total_prometheus":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false, false), nil
+		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false), nil
 	case "unique_samples":
 		return newUniqueSamplesAggrConfig(), nil
 	default:
@@ -845,6 +845,7 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipFlu
 			} else {
 				a.flush(pf, flushTime, cs, false)
 			}
+			flushTime = flushTime.Add(a.interval)
 			for time.Now().After(flushTime) {
 				flushTime = flushTime.Add(a.interval)
 			}

lib/streamaggr/streamaggr_synctest_test.go

@@ -1,5 +1,3 @@
-//go:build synctest
-
 package streamaggr
 
 import (
@@ -485,10 +483,8 @@ foo 3.3
 `, ``, ``, ``, ``}, time.Minute, `foo:1m_count_series 1
 foo:1m_count_series{bar="baz"} 1
 foo:1m_sum_samples 0
-foo:1m_sum_samples 0
 foo:1m_sum_samples 4.3
 foo:1m_sum_samples{bar="baz"} 0
-foo:1m_sum_samples{bar="baz"} 0
 foo:1m_sum_samples{bar="baz"} 2
 foo:5m_by_bar_sum_samples 4.3
 foo:5m_by_bar_sum_samples{bar="baz"} 2
@@ -694,21 +690,29 @@ foo:1m_by_cde_rate_sum{cde="1"} 0.125
 
 	// test rate_sum and rate_avg, when two aggregation intervals are empty
 	f([]string{`
-foo{abc="123", cde="1"} 2
-foo{abc="456", cde="1"} 8
-foo{abc="777", cde="1"} 9 -10
+foo{abc="123", cde="1"} 1
+foo{abc="123", cde="1"} 2 1
+foo{abc="456", cde="1"} 7
+foo{abc="456", cde="1"} 8 1
+foo{abc="777", cde="1"} 8
+foo{abc="777", cde="1"} 9 1
 `, ``, ``, `
-foo{abc="123", cde="1"} 20
+foo{abc="123", cde="1"} 19
+foo{abc="123", cde="1"} 20 1
 foo{abc="456", cde="1"} 26
-foo{abc="777", cde="1"} 27 -10
-`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 0.1
-foo:1m_by_cde_rate_sum{cde="1"} 0.2
+foo{abc="456", cde="1"} 27 1
+foo{abc="777", cde="1"} 27
+foo{abc="777", cde="1"} 28 1
+`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 1
+foo:1m_by_cde_rate_avg{cde="1"} 1
+foo:1m_by_cde_rate_sum{cde="1"} 3
+foo:1m_by_cde_rate_sum{cde="1"} 3
 `, `            
 - interval: 1m
   by: [cde]
   outputs: [rate_sum, rate_avg]
   enable_windows: true
-`, "111111")
+`, "111111111111")
 
 	// rate_sum and rate_avg with duplicated events
 	f([]string{`

lib/streamaggr/streamaggr_timing_test.go

@@ -74,7 +74,7 @@ func newBenchAggregators(outputs []string, pushFunc PushFunc) *Aggregators {
 `, strings.Join(outputsQuoted, ","))
 	a, err := LoadFromData([]byte(config), pushFunc, nil, "some_alias")
 	if err != nil {
-		panic(fmt.Errorf("unexpected error when initializing aggregators: %s", err))
+		panic(fmt.Errorf("unexpected error when initializing aggregators: %w", err))
 	}
 	return a
 }
@@ -133,7 +133,7 @@ func newPerOutputBenchAggregators(outputs []string, pushFunc PushFunc) *Aggregat
 
 	a, err := LoadFromData([]byte(config), pushFunc, nil, "some_alias")
 	if err != nil {
-		panic(fmt.Errorf("unexpected error when initializing aggregators: %s", err))
+		panic(fmt.Errorf("unexpected error when initializing aggregators: %w", err))
 	}
 	return a
 }

lib/streamaggr/total.go

@@ -53,36 +53,30 @@ func (av *totalAggrValue) pushSample(c aggrConfig, sample *pushSample, key strin
 
 func (av *totalAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, isLast bool) {
 	ac := c.(*totalAggrConfig)
-	suffix := ac.getSuffix()
-	// check for stale entries
 	total := av.shared.total + av.total
 	av.total = 0
-	lvs := av.shared.lastValues
-	for lk, lv := range lvs {
+	for lk, lv := range av.shared.lastValues {
 		if ctx.flushTimestamp > lv.deleteDeadline || isLast {
-			delete(lvs, lk)
+			delete(av.shared.lastValues, lk)
 		}
 	}
-	if ac.resetTotalOnFlush {
-		av.shared.total = 0
-	} else if math.Abs(total) >= (1 << 53) {
+	if math.Abs(total) >= (1 << 53) {
 		// It is time to reset the entry, since it starts losing float64 precision
 		av.shared.total = 0
 	} else {
 		av.shared.total = total
 	}
-	ctx.appendSeries(key, suffix, total)
+	ctx.appendSeries(key, ac.getSuffix(), total)
 }
 
 func (av *totalAggrValue) state() any {
 	return av.shared
 }
 
-func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, resetTotalOnFlush, keepFirstSample bool) aggrConfig {
+func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, keepFirstSample bool) aggrConfig {
 	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + ignoreFirstSampleIntervalSecs
 	cfg := &totalAggrConfig{
 		keepFirstSample:           keepFirstSample,
-		resetTotalOnFlush:         resetTotalOnFlush,
 		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
 	}
 	cfg.counterResetsTotal = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_counter_resets_total{%s}`, metricLabels))
@@ -90,8 +84,6 @@ func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleI
 }
 
 type totalAggrConfig struct {
-	resetTotalOnFlush bool
-
 	// Whether to take into account the first sample in new time series when calculating the output value.
 	keepFirstSample bool
 
@@ -117,12 +109,6 @@ func (*totalAggrConfig) getValue(s any) aggrValue {
 }
 
 func (ac *totalAggrConfig) getSuffix() string {
-	if ac.resetTotalOnFlush {
-		if ac.keepFirstSample {
-			return "increase"
-		}
-		return "increase_prometheus"
-	}
 	if ac.keepFirstSample {
 		return "total"
 	}