deployment/docker/rules: fixed expression for vmalert rules

app/vmctl/vm_native.go

@@ -120,17 +120,11 @@ func (p *vmNativeProcessor) do(ctx context.Context, f native.Filter, srcURL, dst
 }
 
 func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcURL, dstURL string, bar barpool.Bar) error {
-	exportReader, err := p.src.ExportPipe(ctx, srcURL, f)
+	reader, err := p.src.ExportPipe(ctx, srcURL, f)
 	if err != nil {
 		return fmt.Errorf("failed to init export pipe: %w", err)
 	}
-	defer func() {
-		// close the export reader on exit, so it doesn't hang at the source
-		// until server-side timeout.
-		_ = exportReader.Close()
-	}()
 
-	reader := io.Reader(exportReader)
 	if p.disablePerMetricRequests {
 		pr := bar.NewProxyReader(reader)
 		if pr != nil {
@@ -140,11 +134,10 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
 	}
 
 	pr, pw := io.Pipe()
-	// make importCh buffered so goroutine won't get stuck if nothing reads from chan
-	importCh := make(chan error, 1)
+	importCh := make(chan error)
 	go func() {
 		importCh <- p.dst.ImportPipe(ctx, dstURL, pr)
-		_ = pr.Close()
+		close(importCh)
 	}()
 
 	w := io.Writer(pw)
@@ -155,17 +148,13 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
 
 	written, err := io.Copy(w, reader)
 	if err != nil {
-		// close the import writer, so the destination doesn't hang until server-side timeout
-		_ = pw.CloseWithError(err)
+		// io.Copy could fail if ImportPipe will fail before and close the pr
+		// so we check if that's the case and to not ignore importErr if it exists.
 		select {
-		// check if the error happened in the ImportPipe
 		case importErr := <-importCh:
 			if importErr != nil {
 				return fmt.Errorf("failed to import %s: %w", p.dst.Addr, importErr)
 			}
-		// or because vmctl has been stopped
-		case <-ctx.Done():
-			return ctx.Err()
 		default:
 		}
 		return fmt.Errorf("failed to write into %q: %w", p.dst.Addr, err)

app/vmctl/vm_native_test.go

@@ -1,15 +1,7 @@
 package main
 
 import (
-	"context"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"sync/atomic"
 	"testing"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/native"
 )
 
 func TestBuildMatchWithFilter_Failure(t *testing.T) {
@@ -72,102 +64,3 @@ func TestBuildMatchWithFilter_Success(t *testing.T) {
 	// metric name has negative regex and metric name is empty
 	f(`{__name__!~".*"}`, "", `{__name__!~".*"}`)
 }
-
-// newExportServer returns a server, which streams chunks of data.
-// It will abort the response if abort is set.
-func newExportServer(t *testing.T, chunks int, abort bool) *httptest.Server {
-	t.Helper()
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		w.WriteHeader(http.StatusOK)
-		// the chunk size is intentionally big to exceed socket buffers. Otherwise, response from export server
-		// could be written into the socket even before import server responded.
-		buf := make([]byte, 128*1024)
-		for range chunks {
-			if _, err := w.Write(buf); err != nil {
-				return
-			}
-		}
-		w.(http.Flusher).Flush()
-		if abort {
-			// http.ErrAbortHandler closes the connection without logging the stack trace
-			panic(http.ErrAbortHandler)
-		}
-	}))
-}
-
-func TestVMNativeProcessorRunSingle_ExportFailureDoesntHang(t *testing.T) {
-	var importsStarted, importsInFlight atomic.Int64
-
-	src := newExportServer(t, 4, true)
-	defer src.Close()
-
-	// The destination reads the request body until it is closed, like vminsert does.
-	dst := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		importsStarted.Add(1)
-		importsInFlight.Add(1)
-		defer importsInFlight.Add(-1)
-		_, _ = io.Copy(io.Discard, r.Body)
-		w.WriteHeader(http.StatusNoContent)
-	}))
-	defer dst.Close()
-
-	p := &vmNativeProcessor{
-		s:   &stats{startTime: time.Now()},
-		src: &native.Client{Addr: src.URL, HTTPClient: &http.Client{}},
-		dst: &native.Client{Addr: dst.URL, HTTPClient: &http.Client{}},
-	}
-
-	const attempts = 3
-	for i := range attempts {
-		if err := p.runSingle(context.Background(), native.Filter{}, src.URL, dst.URL, nil); err == nil {
-			t.Fatalf("expecting non-nil error on attempt %d", i)
-		}
-	}
-
-	if n := importsStarted.Load(); n != attempts {
-		t.Fatalf("unexpected number of import requests; got %d; want %d", n, attempts)
-	}
-
-	// Every failed attempt must abort its import request at the destination.
-	// Otherwise, the requests pile up there for the whole lifetime of the migration.
-	deadline := time.Now().Add(5 * time.Second)
-	for importsInFlight.Load() > 0 && time.Now().Before(deadline) {
-		time.Sleep(10 * time.Millisecond)
-	}
-	if n := importsInFlight.Load(); n != 0 {
-		t.Fatalf("%d import requests are still in flight at the destination after %d failed attempts", n, attempts)
-	}
-}
-
-func TestVMNativeProcessorRunSingle_Success(t *testing.T) {
-	const chunks = 8
-	var got atomic.Int64
-
-	src := newExportServer(t, chunks, false)
-	defer src.Close()
-
-	dst := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		n, _ := io.Copy(io.Discard, r.Body)
-		got.Add(n)
-		w.WriteHeader(http.StatusNoContent)
-	}))
-	defer dst.Close()
-
-	p := &vmNativeProcessor{
-		s:   &stats{startTime: time.Now()},
-		src: &native.Client{Addr: src.URL, HTTPClient: &http.Client{}},
-		dst: &native.Client{Addr: dst.URL, HTTPClient: &http.Client{}},
-	}
-
-	if err := p.runSingle(context.Background(), native.Filter{}, src.URL, dst.URL, nil); err != nil {
-		t.Fatalf("unexpected runSingle() error: %s", err)
-	}
-
-	want := int64(chunks * 128 * 1024)
-	if got.Load() != want {
-		t.Fatalf("unexpected number of bytes at the destination; got %d; want %d", got.Load(), want)
-	}
-	if p.s.bytes != uint64(want) {
-		t.Fatalf("unexpected stats.bytes; got %d; want %d", p.s.bytes, want)
-	}
-}

app/vmselect/netstorage/netstorage.go

@@ -460,8 +460,7 @@ func (pts *packedTimeseries) unpackTo(dst []*sortBlock, tbf *tmpBlocksFile, tr s
 			initUnpackWork(upw, br)
 			upw.unpack(tmpBlock)
 			if upw.err != nil {
-				err = upw.err
-				break
+				return dst, upw.err
 			}
 			samples += len(upw.sb.Timestamps)
 			if *maxSamplesPerSeries > 0 && samples > *maxSamplesPerSeries {
@@ -475,11 +474,7 @@ func (pts *packedTimeseries) unpackTo(dst []*sortBlock, tbf *tmpBlocksFile, tr s
 		}
 		putTmpStorageBlock(tmpBlock)
 		putUnpackWork(upw)
-		if err != nil {
-			for _, sb := range dst {
-				putSortBlock(sb)
-			}
-		}
+
 		return dst, err
 	}
 
@@ -545,11 +540,6 @@ func (pts *packedTimeseries) unpackTo(dst []*sortBlock, tbf *tmpBlocksFile, tr s
 		}
 		putUnpackWork(upw)
 	}
-	if firstErr != nil {
-		for _, sb := range dst {
-			putSortBlock(sb)
-		}
-	}
 
 	return dst, firstErr
 }

app/vmselect/prometheus/prometheus.go

@@ -703,7 +703,7 @@ var labelsDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/
 func SeriesCountHandler(startTime time.Time, w http.ResponseWriter, r *http.Request) error {
 	defer seriesCountDuration.UpdateDuration(startTime)
 
-	deadline := searchutil.GetDeadlineForLabelsAPI(r, startTime)
+	deadline := searchutil.GetDeadlineForStatusRequest(r, startTime)
 	n, err := netstorage.SeriesCount(nil, deadline)
 	if err != nil {
 		return fmt.Errorf("cannot obtain series count: %w", err)

app/vmselect/promql/eval.go

@@ -1072,7 +1072,6 @@ func evalRollupFuncWithSubquery(qt *querytracer.Tracer, ec *EvalConfig, funcName
 	var samplesScannedTotal atomic.Uint64
 	keepMetricNames := getKeepMetricNames(expr)
 	tsw := getTimeseriesByWorkerID()
-	defer putTimeseriesByWorkerID(tsw)
 	seriesByWorkerID := tsw.byWorkerID
 	doParallel(tssSQ, func(tsSQ *timeseries, values []float64, timestamps []int64, workerID uint) ([]float64, []int64) {
 		values, timestamps = removeNanValues(values[:0], timestamps[:0], tsSQ.Values, tsSQ.Timestamps)
@@ -1095,6 +1094,7 @@ func evalRollupFuncWithSubquery(qt *querytracer.Tracer, ec *EvalConfig, funcName
 	for i := range seriesByWorkerID {
 		tss = append(tss, seriesByWorkerID[i].tss...)
 	}
+	putTimeseriesByWorkerID(tsw)
 
 	rowsScannedPerQuery.Update(float64(samplesScannedTotal.Load()))
 	qt.Printf("rollup %s() over %d series returned by subquery: series=%d, samplesScanned=%d", funcName, len(tssSQ), len(tss), samplesScannedTotal.Load())
@@ -1973,7 +1973,6 @@ func evalRollupNoIncrementalAggregate(qt *querytracer.Tracer, funcName string, k
 
 	var samplesScannedTotal atomic.Uint64
 	tsw := getTimeseriesByWorkerID()
-	defer putTimeseriesByWorkerID(tsw)
 	seriesByWorkerID := tsw.byWorkerID
 	seriesLen := rss.Len()
 	err := rss.RunParallel(qt, func(rs *netstorage.Result, workerID uint) error {
@@ -2000,6 +1999,7 @@ func evalRollupNoIncrementalAggregate(qt *querytracer.Tracer, funcName string, k
 	for i := range seriesByWorkerID {
 		tss = append(tss, seriesByWorkerID[i].tss...)
 	}
+	putTimeseriesByWorkerID(tsw)
 
 	rowsScannedPerQuery.Update(float64(samplesScannedTotal.Load()))
 	qt.Printf("samplesScanned=%d", samplesScannedTotal.Load())

app/vmselect/searchutil/searchutil.go

@@ -56,7 +56,7 @@ func GetDeadlineForExport(r *http.Request, startTime time.Time) Deadline {
 	return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxExportDuration")
 }
 
-// GetDeadlineForLabelsAPI returns deadline for the given request to /api/v1/labels, /api/v1/label/.../values, /api/v1/series or /api/v1/series/count
+// GetDeadlineForLabelsAPI returns deadline for the given request to /api/v1/labels, /api/v1/label/.../values or /api/v1/series
 func GetDeadlineForLabelsAPI(r *http.Request, startTime time.Time) Deadline {
 	dMax := maxLabelsAPIDuration.Milliseconds()
 	return getDeadlineWithMaxDuration(r, startTime, dMax, "-search.maxLabelsAPIDuration")

deployment/docker/rules/alerts-vmalert.yml

@@ -126,10 +126,10 @@ groups:
           (
             vmalert_alerting_rules_last_evaluation_samples
               > on(group,file) group_left()
-            (vmalert_group_rule_results_limit * 0.9)
+            min by (group,file) (vmalert_group_rule_results_limit * 0.9)
           )
             and on(group,file)
-          (vmalert_group_rule_results_limit > 0)
+          (min by (group,file) (vmalert_group_rule_results_limit) > 0)
         for: 5m
         labels:
           severity: warning
@@ -144,10 +144,10 @@ groups:
           (
             vmalert_recording_rules_last_evaluation_samples
               > on(group,file) group_left()
-            (vmalert_group_rule_results_limit * 0.9)
+            min by (group,file) (vmalert_group_rule_results_limit * 0.9)
           )
             and on(group,file)
-          (vmalert_group_rule_results_limit > 0)
+          (min by (group,file) (vmalert_group_rule_results_limit) > 0)
         for: 5m
         labels:
           severity: warning

docs/victoriametrics/Articles.md

@@ -117,10 +117,6 @@ See also [case studies](https://docs.victoriametrics.com/victoriametrics/casestu
 * [Claude Code: creating Kubernetes debugging AI Agent for VictoriaMetrics](https://rtfm.co.ua/en/claude-code-creating-kubernetes-debugging-ai-agent-for-victoriametrics/)
 * [OpenTelemetry: OTel Collectors in Kubernetes and VictoriaMetrics Stack integration](https://itnext.io/opentelemetry-otel-collectors-in-kubernetes-and-victoriametrics-stack-integration-d907ed0a15a0)
 * [VictoriaMetrics vs Prometheus: my default, and when I still pick Prometheus](https://jorijn.com/en/blog/victoriametrics-vs-prometheus/)
-* [LiteLLM: Monitoring with VictoriaMetrics – Alerts and Grafana](https://rtfm.co.ua/en/litellm-monitoring-with-victoriametrics-alerts-and-grafana/)
-* [LiteLLM: AI Gateway on Kubernetes and Metrics in VictoriaMetrics](https://rtfm.co.ua/en/litellm-ai-gateway-on-kubernetes-and-metrics-in-victoriametrics/)
-* [LiteLLM: Metrics, Traces, and VictoriaMetrics Stack Integration](https://rtfm.co.ua/en/litellm-metrics-traces-and-victoriametrics-stack-integration/)
-* [llama.cpp: Metrics and Monitoring with VictoriaMetrics](https://rtfm.co.ua/en/llama-cpp-metrics-and-monitoring-with-victoriametrics/)
 
 ## Third-party articles and slides about VictoriaLogs
 

docs/victoriametrics/changelog/CHANGELOG.md

@@ -33,11 +33,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix infinite loop in the OpenTelemetry Firehose ingestion endpoint (`/opentelemetry/api/v1/push`) when receiving a malformed record with an incomplete varint in the `data` field. Previously this caused the goroutine to spin forever, permanently consuming CPU until the process was restarted.
 * BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/): reuse connections to `-remoteWrite.url` when writing the results of recording rules and alerts. Previously every series was sent over a new connection, which left a lot of sockets in `TIME_WAIT` state and could exhaust the ephemeral port range. The number of idle connections can be tuned via the new `-remoteWrite.maxIdleConnections` command-line flag. Thanks @evkuzin for contribution.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): prevent process crash in `sort_by_label_numeric()` and `sort_by_label_numeric_desc()` when a label value contains a number with 309 or more digits. See [#11423](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11423).
-* BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly apply default query timeout to `/api/v1/series/count` requests. It used `-search.maxStatusRequestDuration` flag value instead of `-search.maxLabelsAPIDuration`. See [#11422](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11422).
-* BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly re-use memory if query aggregation returns error. See [#11426](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11426).
 * BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fail the query request directly when there is not enough disk space to store temporary search results. Previously, such queries could lead to vmselect crash. See [#4688](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4688).
-* BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): consistently re-use memory during storage blocks unpacking on parsing storage block error. See [#11421](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11421).
-* BUGFIX: [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): properly release the export and import requests during migration requests fails in [vm-native mode](https://docs.victoriametrics.com/victoriametrics/vmctl/#migrating-data-from-victoriametrics). Previously, failed export/import requests could have left hanging at the source or the destination. The fix is supposed to improve the resiliency of vmctl during long-running migrations.
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)
 

lib/promscrape/scrapework.go

@@ -429,14 +429,12 @@ func (sw *scrapeWork) needStreamParseMode(responseSize int) bool {
 // getTargetResponse() fetches response from sw target in the same way as when scraping the target.
 func (sw *scrapeWork) getTargetResponse() ([]byte, error) {
 	cb := chunkedbuffer.Get()
+	defer chunkedbuffer.Put(cb)
 
 	isGzipped, err := sw.ReadData(cb)
 	if err != nil {
 		return nil, err
 	}
-	// in case of error buffer cannot be returned back to the pool
-	// See https://pkg.go.dev/net/http#RoundTripper
-	defer chunkedbuffer.Put(cb)
 
 	var bb bytesutil.ByteBuffer
 	err = sw.readFromBuffer(&bb, cb, isGzipped)
@@ -468,8 +466,8 @@ func (sw *scrapeWork) scrapeInternal(scrapeTimestamp, realTimestamp int64) error
 	body := leveledbytebufferpool.Get(sw.prevBodyLen)
 	if err == nil {
 		err = sw.readFromBuffer(body, cb, isGzipped)
-		chunkedbuffer.Put(cb)
 	}
+	chunkedbuffer.Put(cb)
 
 	bodyLen := len(body.B)
 	sw.prevBodyLen = bodyLen