deployment/docker/rules: fixed expression for vmalert rules

Previously, vmselect may crash if there is not enough free disk space to store tmp files from vmstorage. vmselect stores data blocks received from vmstorage if response size exceeds in-memory limit.
 It reduced vmselect availability, because it makes impossible to serve small queries. And huge query may crash vmselect for all users.

 So this commit adds a error check instead. If there is not enough disk space, vmselect will stop query execution and return error back to user.

fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4688

README.md

@@ -1,6 +1,6 @@
 # VictoriaMetrics
 
-[![Latest Release](https://img.shields.io/github/v/release/VictoriaMetrics/VictoriaMetrics?sort=semver&label=&filter=!*-victorialogs&logo=github&labelColor=gray&color=gray&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Freleases%2Flatest)](https://github.com/VictoriaMetrics/VictoriaMetrics/releases)
+[![Latest Release](https://img.shields.io/github/v/release/VictoriaMetrics/VictoriaMetrics?logo=github&labelColor=gray&color=gray&label=Release)](https://github.com/VictoriaMetrics/VictoriaMetrics/releases)
 [![Docker Pulls](https://img.shields.io/docker/pulls/victoriametrics/victoria-metrics?label=&logo=docker&logoColor=white&labelColor=2496ED&color=2496ED&link=https%3A%2F%2Fhub.docker.com%2Fr%2Fvictoriametrics%2Fvictoria-metrics)](https://hub.docker.com/u/victoriametrics)
 [![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/build.yml/badge.svg?branch=master&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Factions)](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/build.yml)
 [![License](https://img.shields.io/github/license/VictoriaMetrics/VictoriaMetrics?labelColor=green&label=&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Fblob%2Fmaster%2FLICENSE)](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/LICENSE)

app/victoria-metrics/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
@@ -34,9 +35,10 @@ var (
 		"This can be changed with -promscrape.config.strictParse=false command-line flag")
 	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
 		"By default there are no limits on samples ingestion rate.")
-	vmselectMaxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
-		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
-		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
+	vmselectMaxConcurrentRequests = flagutil.NewIntWithDynamicDefault("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "vmselect.getDefaultMaxConcurrentRequests()",
+		"The maximum number of concurrent search requests. "+
+			"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
+			"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
 	vmselectMaxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
 		"limit is reached; see also -search.maxQueryDuration")
 )
@@ -90,7 +92,9 @@ func main() {
 	}
 	logger.Infof("starting VictoriaMetrics at %q...", listenAddrs)
 	startTime := time.Now()
+
 	vmstorage.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration, promql.ResetRollupResultCacheIfNeeded)
+	appmetrics.MustCreateUncleanShutdownMarker(vmstorage.DataPath())
 	vmselect.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration)
 	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
 	vminsert.Init()
@@ -120,6 +124,7 @@ func main() {
 
 	vmstorage.Stop()
 	vmselect.Stop()
+	appmetrics.MustRemoveUncleanShutdownMarker(vmstorage.DataPath())
 
 	logger.Infof("the VictoriaMetrics has been stopped in %.3f seconds", time.Since(startTime).Seconds())
 }

app/vmagent/remotewrite/remotewrite.go

@@ -15,6 +15,7 @@ import (
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/cespare/xxhash/v2"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
@@ -62,9 +63,10 @@ var (
 		"See also -remoteWrite.maxDiskUsagePerURL and -remoteWrite.disableOnDiskQueue")
 	keepDanglingQueues = flag.Bool("remoteWrite.keepDanglingQueues", false, "Keep persistent queues contents at -remoteWrite.tmpDataPath in case there are no matching -remoteWrite.url. "+
 		"Useful when -remoteWrite.url is changed temporarily and persistent queue files will be needed later on.")
-	queues = flagutil.NewArrayInt("remoteWrite.queues", cgroup.AvailableCPUs()*2, "The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
-		"isn't enough for sending high volume of collected data to remote storage. "+
-		"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
+	queues = flagutil.NewArrayIntWithDynamicDefault("remoteWrite.queues", cgroup.AvailableCPUs()*2, "2*cgroup.AvailableCPUs()",
+		"The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
+			"isn't enough for sending high volume of collected data to remote storage. "+
+			"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
 	inmemoryQueues = flagutil.NewArrayInt("remoteWrite.inmemoryQueues", 0, "The number of additional workers per each -remoteWrite.url, which send only recently ingested data from the in-memory queue, "+
 		"while the file-based queue at -remoteWrite.tmpDataPath is drained by workers configured via -remoteWrite.queues. "+
 		"This reduces delivery lag for fresh samples when the file-based queue contains a backlog accumulated during remote storage outages.")
@@ -233,6 +235,7 @@ func Init() {
 	initStreamAggrConfigGlobal()
 
 	initRemoteWriteCtxs(*remoteWriteURLs)
+	appmetrics.MustCreateUncleanShutdownMarker(*tmpDataPath)
 
 	disableOnDiskQueues := []bool(*disableOnDiskQueue)
 	disableOnDiskQueueAny = slices.Contains(disableOnDiskQueues, true)
@@ -391,6 +394,8 @@ func Stop() {
 	if sl := dailySeriesLimiter; sl != nil {
 		sl.MustStop()
 	}
+
+	appmetrics.MustRemoveUncleanShutdownMarker(*tmpDataPath)
 }
 
 // PushDropSamplesOnFailure pushes wr to the configured remote storage systems set via -remoteWrite.url

app/vmalert/remotewrite/debug_client.go

@@ -36,6 +36,13 @@ func NewDebugClient() (*DebugClient, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for -remoteWrite.url=%q: %w", *addr, err)
 	}
+	tr.IdleConnTimeout = *idleConnectionTimeout
+	// DebugClient sends every series in a separate request, so it needs more idle
+	// connections than the two http.DefaultTransport keeps per host.
+	tr.MaxIdleConnsPerHost = *maxIdleConnections
+	if tr.MaxIdleConns != 0 && tr.MaxIdleConns < tr.MaxIdleConnsPerHost {
+		tr.MaxIdleConns = tr.MaxIdleConnsPerHost
+	}
 	c := &DebugClient{
 		c: &http.Client{
 			Timeout:   *sendTimeout,

app/vmalert/remotewrite/debug_client_conns_test.go

@@ -0,0 +1,45 @@
+package remotewrite
+
+import (
+	"net/http"
+	"testing"
+)
+
+// TestDebugClient_IdleConns makes sure DebugClient keeps enough idle connections
+// to -remoteWrite.url. Every series is pushed in a separate request, so with the
+// two idle connections per host of http.DefaultTransport most of the concurrent
+// requests would open a new connection and leave a socket in TIME_WAIT state.
+func TestDebugClient_IdleConns(t *testing.T) {
+	f := func(maxIdle int) {
+		t.Helper()
+
+		oldAddr, oldMaxIdle := *addr, *maxIdleConnections
+		*addr, *maxIdleConnections = "http://localhost:8428", maxIdle
+		defer func() {
+			*addr, *maxIdleConnections = oldAddr, oldMaxIdle
+		}()
+
+		client, err := NewDebugClient()
+		if err != nil {
+			t.Fatalf("failed to create debug client: %s", err)
+		}
+		tr, ok := client.c.Transport.(*http.Transport)
+		if !ok {
+			t.Fatalf("unexpected transport type %T", client.c.Transport)
+		}
+		if tr.MaxIdleConnsPerHost != maxIdle {
+			t.Fatalf("unexpected MaxIdleConnsPerHost; got %d; want %d", tr.MaxIdleConnsPerHost, maxIdle)
+		}
+		if tr.MaxIdleConns != 0 && tr.MaxIdleConns < maxIdle {
+			t.Fatalf("MaxIdleConns=%d is lower than MaxIdleConnsPerHost=%d", tr.MaxIdleConns, maxIdle)
+		}
+		if tr.IdleConnTimeout != *idleConnectionTimeout {
+			t.Fatalf("unexpected IdleConnTimeout; got %s; want %s", tr.IdleConnTimeout, *idleConnectionTimeout)
+		}
+	}
+
+	f(100)
+
+	// the number of idle connections must be raised together with the total limit
+	f(1000)
+}

app/vmalert/remotewrite/init.go

@@ -34,10 +34,12 @@ var (
 	bearerTokenFile = flag.String("remoteWrite.bearerTokenFile", "", "Optional path to bearer token file to use for -remoteWrite.url.")
 
 	idleConnectionTimeout = flag.Duration("remoteWrite.idleConnTimeout", 50*time.Second, `Defines a duration for idle (keep-alive connections) to exist. Consider settings this value less to the value of "-http.idleConnTimeout". It must prevent possible "write: broken pipe" and "read: connection reset by peer" errors.`)
+	maxIdleConnections    = flag.Int("remoteWrite.maxIdleConnections", 100, `Defines the number of idle (keep-alive connections) to -remoteWrite.url for the vmalert-tool debug writer, which sends every series in a separate request. Too low a value may result in a high number of sockets in TIME_WAIT state.`)
 
-	maxQueueSize  = flag.Int("remoteWrite.maxQueueSize", defaultMaxQueueSize, "Defines the max number of pending datapoints to remote write endpoint")
-	maxBatchSize  = flag.Int("remoteWrite.maxBatchSize", defaultMaxBatchSize, "Defines max number of timeseries to be flushed at once")
-	concurrency   = flag.Int("remoteWrite.concurrency", defaultConcurrency, "Defines number of writers for concurrent writing into remote write endpoint. Default value depends on the number of available CPU cores.")
+	maxQueueSize = flag.Int("remoteWrite.maxQueueSize", defaultMaxQueueSize, "Defines the max number of pending datapoints to remote write endpoint")
+	maxBatchSize = flag.Int("remoteWrite.maxBatchSize", defaultMaxBatchSize, "Defines max number of timeseries to be flushed at once")
+	concurrency  = flagutil.NewIntWithDynamicDefault("remoteWrite.concurrency", defaultConcurrency, "2*cgroup.AvailableCPUs()",
+		"Defines number of writers for concurrent writing into remote write endpoint. Default value depends on the number of available CPU cores.")
 	flushInterval = flag.Duration("remoteWrite.flushInterval", defaultFlushInterval, "Defines interval of flushes to remote write endpoint")
 
 	tlsInsecureSkipVerify = flag.Bool("remoteWrite.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -remoteWrite.url")

app/vmalert/rule/alerting.go

@@ -437,7 +437,7 @@ const resolvedRetention = 15 * time.Minute
 
 // exec executes AlertingRule expression via the given Querier.
 // Based on the Querier results AlertingRule maintains notifier.Alerts
-func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) ([]prompb.TimeSeries, error) {
+func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := ar.q.Query(ctx, ar.Expr, ts)
 	curState := StateEntry{
@@ -546,15 +546,6 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int, getRe
 		ar.alerts[alertID] = a
 		ar.logDebugf(ts, a, "created in state PENDING")
 	}
-	// try to restore alerts state from remoteRead if necessary
-	if getRemoteReadQuerier != nil {
-		rr := getRemoteReadQuerier(ar.Debug)
-		err := ar.restore(ctx, rr, ts)
-		// do not break the current evaluation if restore request fails
-		if err != nil {
-			logger.Errorf("error while restoring ruleState for group %q(file %q) rule %q: %s", ar.GroupName, ar.File, ar.Name, err)
-		}
-	}
 	var numActivePending int
 	var tss []prompb.TimeSeries
 	for h, a := range ar.alerts {
@@ -808,7 +799,7 @@ func firingAlertStaleTimeSeries(ls map[string]string, timestamp int64) []prompb.
 // restore restores the value of ActiveAt field for active alerts,
 // based on previously written time series `alertForStateMetricName`.
 // Only rules with For > 0 can be restored.
-func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts time.Time) error {
+func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts time.Time, lookback time.Duration) error {
 	if ar.For < 1 {
 		return nil
 	}
@@ -834,9 +825,11 @@ func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts ti
 	}
 	// use `default_rollup()` instead of `last_over_time()` here to accounts for possible staleness markers
 	expr := fmt.Sprintf("default_rollup(%s{%s%s}[%ds])",
-		alertForStateMetricName, nameStr, labelsFilter, int(remoteReadLookBack.Seconds()))
+		alertForStateMetricName, nameStr, labelsFilter, int(lookback.Seconds()))
 
-	res, _, err := q.Query(ctx, expr, ts)
+	// query ALERTS_FOR_STATE at `ts-1s` instead `ts` to avoid retrieving data written in the current run,
+	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10335
+	res, _, err := q.Query(ctx, expr, ts.Add(-1*time.Second))
 	if err != nil {
 		return fmt.Errorf("failed to execute restore query %q: %w ", expr, err)
 	}
@@ -846,6 +839,9 @@ func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts ti
 		return nil
 	}
 
+	ar.alertsMu.Lock()
+	defer ar.alertsMu.Unlock()
+
 	for _, series := range res.Data {
 		series.DelLabel("__name__")
 		labelSet := make(map[string]string, len(series.Labels))

app/vmalert/rule/alerting_synctest_test.go

@@ -44,7 +44,7 @@ func TestAlertingRule_ActiveAtPreservedInAnnotations(t *testing.T) {
 
 		// First execution - creates new alert
 		ts1 := time.Now()
-		_, err := ar.exec(context.TODO(), ts1, 0, nil)
+		_, err := ar.exec(context.TODO(), ts1, 0)
 		if err != nil {
 			t.Fatalf("unexpected error on first exec: %s", err)
 		}
@@ -71,7 +71,7 @@ func TestAlertingRule_ActiveAtPreservedInAnnotations(t *testing.T) {
 		// sleep is non-blocking thanks to synctest
 		time.Sleep(2 * time.Second)
 		ts2 := time.Now()
-		_, err = ar.exec(context.TODO(), ts2, 0, nil)
+		_, err = ar.exec(context.TODO(), ts2, 0)
 		if err != nil {
 			t.Fatalf("unexpected error on second exec: %s", err)
 		}

app/vmalert/rule/alerting_test.go

@@ -229,7 +229,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		for i, step := range steps {
 			fq.Reset()
 			fq.Add(step...)
-			tss, err := rule.exec(context.TODO(), ts, 0, nil)
+			tss, err := rule.exec(context.TODO(), ts, 0)
 			if err != nil {
 				t.Fatalf("unexpected error: %s", err)
 			}
@@ -824,11 +824,10 @@ func TestAlertingRuleExecRange(t *testing.T) {
 func TestGroup_Restore(t *testing.T) {
 	defaultTS := time.Now()
 	fqr := &datasource.FakeQuerierWithRegistry{}
-	f := func(rules []config.Rule, expAlerts map[uint64]*notifier.Alert, expNotificationNum int) {
+	fn := func(rules []config.Rule, expAlerts map[uint64]*notifier.Alert) {
 		t.Helper()
 		defer fqr.Reset()
-		fn, cleanup := notifier.InitFakeNotifier()
-		defer cleanup()
+
 		fg := NewGroup(config.Group{Name: "TestRestore", Rules: rules}, fqr, time.Second, nil)
 		fg.Init()
 		wg := sync.WaitGroup{}
@@ -858,8 +857,8 @@ func TestGroup_Restore(t *testing.T) {
 			if !ok {
 				t.Fatalf("expected to have key %d", key)
 			}
-			if got.State != exp.State {
-				t.Fatalf("expected state %d; got %d", exp.State, got.State)
+			if got.State != notifier.StatePending {
+				t.Fatalf("expected state %d; got %d", notifier.StatePending, got.State)
 			}
 			if got.ActiveAt != exp.ActiveAt {
 				t.Fatalf("expected ActiveAt %v; got %v", exp.ActiveAt, got.ActiveAt)
@@ -868,9 +867,6 @@ func TestGroup_Restore(t *testing.T) {
 				t.Fatalf("expected alertname %q; got %q", exp.Name, got.Name)
 			}
 		}
-		if fn.GetCounter() != expNotificationNum {
-			t.Fatalf("expected %d notifications; got %d", expNotificationNum, fn.GetCounter())
-		}
 	}
 
 	stateMetric := func(name string, value time.Time, labels ...string) datasource.Metric {
@@ -882,30 +878,28 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert, no previous state
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
 				ActiveAt: defaultTS,
-				State:    notifier.StatePending,
 			},
-		}, 0)
+		})
 
 	// one active alert with state restore
 	ts := time.Now().Truncate(time.Hour)
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
 		stateMetric("foo", ts))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
-		}, 1)
+		})
 
 	// one rule, two active alerts, one with state restored
 	ts = time.Now().Truncate(time.Hour)
@@ -915,7 +909,7 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
 		// only env=prod has state metric, so only it will have state restore
 		stateMetric("foo", ts, "env", "prod"))
-	f(
+	fn(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
 		},
@@ -923,14 +917,12 @@ func TestGroup_Restore(t *testing.T) {
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
 				ActiveAt: defaultTS,
-				State:    notifier.StatePending,
 			},
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "prod"}): {
 				Name:     "foo",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
-		}, 1)
+		})
 
 	// two rules, two active alerts, one with state restored
 	ts = time.Now().Truncate(time.Hour)
@@ -938,7 +930,7 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set("bar", metricWithValueAndLabels(t, 0, "__name__", "bar"))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
 		stateMetric("bar", ts))
-	f(
+	fn(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
 			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
@@ -947,14 +939,12 @@ func TestGroup_Restore(t *testing.T) {
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
 				ActiveAt: defaultTS,
-				State:    notifier.StatePending,
 			},
 			hash(map[string]string{alertNameLabel: "bar", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "bar",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
-		}, 1)
+		})
 
 	// two rules, two active alerts, two with state restored
 	ts = time.Now().Truncate(time.Hour)
@@ -964,68 +954,63 @@ func TestGroup_Restore(t *testing.T) {
 		stateMetric("foo", ts))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
 		stateMetric("bar", ts))
-	f(
+	fn(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
-			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Hour)},
+			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
 		},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
 			hash(map[string]string{alertNameLabel: "bar", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "bar",
 				ActiveAt: ts,
-				State:    notifier.StatePending,
 			},
-		}, 1)
+		})
 
 	// one active alert but wrong state restore
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertname="bar",alertgroup="TestRestore"}[3600s])`,
 		stateMetric("wrong alert", ts))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
 				ActiveAt: defaultTS,
-				State:    notifier.StatePending,
 			},
-		}, 0)
+		})
 
 	// one active alert with labels
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev"))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
-		}, 1)
+		})
 
 	// one active alert with restore labels mismatch
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev", "team", "foo"))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
 				ActiveAt: defaultTS,
-				State:    notifier.StatePending,
 			},
-		}, 0)
+		})
 
 	// two active alerts with dynamic labels and restore
 	ts = time.Now().Truncate(time.Hour)
@@ -1035,20 +1020,18 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set("foo",
 		metricWithValueAndLabels(t, 0, "__name__", "foo", "env", "dev"),
 		metricWithValueAndLabels(t, 0, "__name__", "foo", "env", "prod"))
-	f(
+	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "{{$labels.env}}"}, For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
 				ActiveAt: ts,
-				State:    notifier.StateFiring,
 			},
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "prod"}): {
 				Name:     "foo",
 				ActiveAt: ts.Add(time.Second),
-				State:    notifier.StateFiring,
 			},
-		}, 2)
+		})
 }
 
 func TestAlertingRule_Exec_Negative(t *testing.T) {
@@ -1061,14 +1044,14 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 	// label `job` will be overridden by rule extra label, the original value will be reserved by "exported_job"
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "baz"))
-	_, err := ar.exec(context.TODO(), time.Now(), 0, nil)
+	_, err := ar.exec(context.TODO(), time.Now(), 0)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	// label `__name__` will be omitted and get duplicated results here
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo_1", "job", "bar"))
-	_, err = ar.exec(context.TODO(), time.Now(), 0, nil)
+	_, err = ar.exec(context.TODO(), time.Now(), 0)
 	if !errors.Is(err, errDuplicate) {
 		t.Fatalf("expected to have %s error; got %s", errDuplicate, err)
 	}
@@ -1077,7 +1060,7 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 
 	expErr := "connection reset by peer"
 	fq.SetErr(errors.New(expErr))
-	_, err = ar.exec(context.TODO(), time.Now(), 0, nil)
+	_, err = ar.exec(context.TODO(), time.Now(), 0)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}
@@ -1100,7 +1083,7 @@ func TestAlertingRuleLimit_Failure(t *testing.T) {
 		fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "bar", "job"))
 
 		timestamp := time.Now()
-		_, err := ar.exec(context.TODO(), timestamp, limit, nil)
+		_, err := ar.exec(context.TODO(), timestamp, limit)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -1128,7 +1111,7 @@ func TestAlertingRuleLimit_Success(t *testing.T) {
 		fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "bar", "job"))
 
 		timestamp := time.Now()
-		_, err := ar.exec(context.TODO(), timestamp, limit, nil)
+		_, err := ar.exec(context.TODO(), timestamp, limit)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -1157,7 +1140,7 @@ func TestAlertingRule_Template(t *testing.T) {
 		fq.SetPartialResponse(isResponsePartial)
 
 		ts := time.Unix(3600, 0)
-		if _, err := rule.exec(context.TODO(), ts, 0, nil); err != nil {
+		if _, err := rule.exec(context.TODO(), ts, 0); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		for hash, expAlert := range alertsExpected {
@@ -1456,7 +1439,7 @@ func TestAlertingRuleExec_Partial(t *testing.T) {
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
 
 	ts := time.Now()
-	_, err := ar.exec(context.TODO(), ts, 0, nil)
+	_, err := ar.exec(context.TODO(), ts, 0)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -1488,7 +1471,7 @@ func TestAlertingRule_QueryTemplateInLabels(t *testing.T) {
 	fq.Add(metricWithValueAndLabels(t, 1, "device", "sda1"))
 
 	ts := time.Now()
-	_, err := ar.exec(context.TODO(), ts, 0, nil)
+	_, err := ar.exec(context.TODO(), ts, 0)
 	if err != nil {
 		t.Fatalf("unexpected error with query template in labels: %s", err)
 	}

app/vmalert/rule/group.go

@@ -222,6 +222,29 @@ func (g *Group) CreateID() uint64 {
 	return hash.Sum64()
 }
 
+// restore restores alerts state for group rules
+func (g *Group) restore(ctx context.Context, qb datasource.QuerierBuilder, ts time.Time, lookback time.Duration) error {
+	for _, rule := range g.Rules {
+		ar, ok := rule.(*AlertingRule)
+		if !ok {
+			continue
+		}
+		if ar.For < 1 {
+			continue
+		}
+		q := qb.BuildWithParams(datasource.QuerierParams{
+			EvaluationInterval: g.Interval,
+			QueryParams:        g.Params,
+			Headers:            g.Headers,
+			Debug:              ar.Debug,
+		})
+		if err := ar.restore(ctx, q, ts, lookback); err != nil {
+			return fmt.Errorf("error while restoring rule %q: %w", rule, err)
+		}
+	}
+	return nil
+}
+
 // updateWith updates existing group with
 // passed group object. This function ignores group
 // evaluation interval change. It supposed to be updated
@@ -372,7 +395,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 
 	g.infof("started")
 
-	eval := func(ctx context.Context, ts time.Time, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) {
+	eval := func(ctx context.Context, ts time.Time) time.Time {
 		g.metrics.iterationTotal.Inc()
 
 		start := time.Now()
@@ -382,13 +405,13 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 			g.mu.Lock()
 			g.LastEvaluation = start
 			g.mu.Unlock()
-			return
+			return ts
 		}
 
 		resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
 		// adjust request timestamp using evalDelay and evalAlignment if necessary
 		ts = g.adjustReqTimestamp(ts)
-		errs := e.execConcurrently(ctx, g.Rules, ts, g.Concurrency, resolveDuration, g.Limit, getRemoteReadQuerier)
+		errs := e.execConcurrently(ctx, g.Rules, ts, g.Concurrency, resolveDuration, g.Limit)
 		for err := range errs {
 			if err != nil {
 				logger.Errorf("group %q (file=%q): %s", g.Name, g.File, err)
@@ -398,6 +421,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 		g.mu.Lock()
 		g.LastEvaluation = start
 		g.mu.Unlock()
+		return ts
 	}
 
 	evalCtx, cancel := context.WithCancel(ctx)
@@ -412,19 +436,16 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 	t := time.NewTicker(g.Interval)
 	defer t.Stop()
 
-	var getRemoteReadQuerier func(enableDebug bool) datasource.Querier
+	realEvalTS := eval(evalCtx, evalTS)
+
+	// restore the rules state after the first evaluation
+	// so only active alerts can be restored.
 	if rr != nil {
-		getRemoteReadQuerier = func(enableDebug bool) datasource.Querier {
-			return rr.BuildWithParams(datasource.QuerierParams{
-				EvaluationInterval: g.Interval,
-				QueryParams:        g.Params,
-				Headers:            g.Headers,
-				Debug:              enableDebug,
-			})
+		err := g.restore(ctx, rr, realEvalTS, *remoteReadLookBack)
+		if err != nil {
+			logger.Errorf("error while restoring ruleState for group %q (file=%q): %s", g.Name, g.File, err)
 		}
 	}
-	// pass getRemoteReadQuerier to the first evaluation, so it can be used for restoring alert states
-	eval(evalCtx, evalTS, getRemoteReadQuerier)
 
 	for {
 		select {
@@ -475,7 +496,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 				g.metrics.iterationMissed.Inc()
 			}
 
-			eval(evalCtx, evalTS, nil)
+			eval(evalCtx, evalTS)
 		}
 	}
 }
@@ -646,7 +667,7 @@ func (g *Group) ExecOnce(ctx context.Context, rw remotewrite.RWClient, evalTS ti
 		return nil
 	}
 	resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
-	return e.execConcurrently(ctx, g.Rules, evalTS, g.Concurrency, resolveDuration, g.Limit, nil)
+	return e.execConcurrently(ctx, g.Rules, evalTS, g.Concurrency, resolveDuration, g.Limit)
 }
 
 type rangeIterator struct {
@@ -720,12 +741,12 @@ type executor struct {
 }
 
 // execConcurrently executes rules concurrently if concurrency>1
-func (e *executor) execConcurrently(ctx context.Context, rules []Rule, ts time.Time, concurrency int, resolveDuration time.Duration, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) chan error {
+func (e *executor) execConcurrently(ctx context.Context, rules []Rule, ts time.Time, concurrency int, resolveDuration time.Duration, limit int) chan error {
 	res := make(chan error, len(rules))
 	if concurrency == 1 {
 		// fast path
 		for _, rule := range rules {
-			res <- e.exec(ctx, rule, ts, resolveDuration, limit, getRemoteReadQuerier)
+			res <- e.exec(ctx, rule, ts, resolveDuration, limit)
 		}
 		close(res)
 		return res
@@ -738,7 +759,7 @@ func (e *executor) execConcurrently(ctx context.Context, rules []Rule, ts time.T
 			rule := rules[i]
 			sem <- struct{}{}
 			wg.Go(func() {
-				res <- e.exec(ctx, rule, ts, resolveDuration, limit, getRemoteReadQuerier)
+				res <- e.exec(ctx, rule, ts, resolveDuration, limit)
 				<-sem
 			})
 		}
@@ -755,10 +776,10 @@ var (
 	execErrors = metrics.NewCounter(`vmalert_execution_errors_total`)
 )
 
-func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDuration time.Duration, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) error {
+func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDuration time.Duration, limit int) error {
 	execTotal.Inc()
 
-	tss, err := r.exec(ctx, ts, limit, getRemoteReadQuerier)
+	tss, err := r.exec(ctx, ts, limit)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
 			// the context can be cancelled on graceful shutdown

app/vmalert/rule/group_test.go

@@ -521,7 +521,7 @@ func TestFaultyNotifier(t *testing.T) {
 	defer cancel()
 
 	go func() {
-		_ = e.exec(ctx, r, time.Now(), 0, 10, nil)
+		_ = e.exec(ctx, r, time.Now(), 0, 10)
 	}()
 
 	tn := time.Now()
@@ -553,7 +553,7 @@ func TestFaultyRW(t *testing.T) {
 		Rw: &remotewrite.Client{},
 	}
 
-	err := e.exec(context.Background(), r, time.Now(), 0, 10, nil)
+	err := e.exec(context.Background(), r, time.Now(), 0, 10)
 	if err == nil {
 		t.Fatalf("expected to get an error from faulty RW client, got nil instead")
 	}

app/vmalert/rule/recording.go

@@ -184,7 +184,7 @@ func (rr *RecordingRule) execRange(ctx context.Context, start, end time.Time) ([
 }
 
 // exec executes RecordingRule expression via the given Querier.
-func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int, _ func(enableDebug bool) datasource.Querier) ([]prompb.TimeSeries, error) {
+func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := rr.q.Query(ctx, rr.Expr, ts)
 	curState := StateEntry{

app/vmalert/rule/recording_test.go

@@ -52,7 +52,7 @@ func TestRecordingRule_Exec(t *testing.T) {
 			rule.state = &ruleState{
 				entries: make([]StateEntry, 10),
 			}
-			tss, err := rule.exec(context.TODO(), ts, 0, nil)
+			tss, err := rule.exec(context.TODO(), ts, 0)
 			if err != nil {
 				t.Fatalf("fail to test rule %s: unexpected error: %s", rule.Name, err)
 			}
@@ -358,7 +358,7 @@ func TestRecordingRuleLimit_Failure(t *testing.T) {
 		}
 		rule.q = fq
 
-		_, err := rule.exec(context.TODO(), time.Now(), limit, nil)
+		_, err := rule.exec(context.TODO(), time.Now(), limit)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -394,7 +394,7 @@ func TestRecordingRuleLimit_Success(t *testing.T) {
 		}
 		rule.q = fq
 
-		_, err := rule.exec(context.TODO(), time.Now(), limit, nil)
+		_, err := rule.exec(context.TODO(), time.Now(), limit)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -422,7 +422,7 @@ func TestRecordingRuleExec_Negative(t *testing.T) {
 	expErr := "connection reset by peer"
 	fq.SetErr(errors.New(expErr))
 	rr.q = fq
-	_, err := rr.exec(context.TODO(), time.Now(), 0, nil)
+	_, err := rr.exec(context.TODO(), time.Now(), 0)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}
@@ -437,7 +437,7 @@ func TestRecordingRuleExec_Negative(t *testing.T) {
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "foo"))
 	fq.Add(metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "bar"))
 
-	_, err = rr.exec(context.TODO(), time.Now(), 0, nil)
+	_, err = rr.exec(context.TODO(), time.Now(), 0)
 	if err != nil {
 		t.Fatalf("cannot execute recording rule: %s", err)
 	}
@@ -479,7 +479,7 @@ func TestRecordingRuleExec_Partial(t *testing.T) {
 	}
 	rule.Debug = true
 	rule.q = fq
-	got, err := rule.exec(context.TODO(), ts, 0, nil)
+	got, err := rule.exec(context.TODO(), ts, 0)
 	want := []prompb.TimeSeries{
 		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompb.Label{
 			{

app/vmalert/rule/rule.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/VictoriaMetrics/metrics"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -28,7 +27,7 @@ type Rule interface {
 	ToAPI() ApiRule
 	// exec executes the rule with given context at the given timestamp and limit.
 	// returns an err if number of resulting time series exceeds the limit.
-	exec(ctx context.Context, ts time.Time, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) ([]prompb.TimeSeries, error)
+	exec(ctx context.Context, ts time.Time, limit int) ([]prompb.TimeSeries, error)
 	// execRange executes the rule on the given time range.
 	execRange(ctx context.Context, start, end time.Time) ([]prompb.TimeSeries, error)
 	// updateWith performs modification of current Rule

app/vmselect/netstorage/netstorage.go

@@ -20,6 +20,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
@@ -30,11 +31,12 @@ var (
 	maxSamplesPerSeries = flag.Int("search.maxSamplesPerSeries", 30e6, "The maximum number of raw samples a single query can scan per each time series. This option allows limiting memory usage")
 	maxSamplesPerQuery  = flag.Int("search.maxSamplesPerQuery", 1e9, "The maximum number of raw samples a single query can process across all time series. "+
 		"This protects from heavy queries, which select unexpectedly high number of raw samples. See also -search.maxSamplesPerSeries")
-	maxWorkersPerQuery = flag.Int("search.maxWorkersPerQuery", defaultMaxWorkersPerQuery, "The maximum number of CPU cores a single query can use. "+
-		"The default value should work good for most cases. "+
-		"The flag can be set to lower values for improving performance of big number of concurrently executed queries. "+
-		"The flag can be set to bigger values for improving performance of heavy queries, which scan big number of time series (>10K) and/or big number of samples (>100M). "+
-		"There is no sense in setting this flag to values bigger than the number of CPU cores available on the system")
+	maxWorkersPerQuery = flagutil.NewIntWithDynamicDefault("search.maxWorkersPerQuery", defaultMaxWorkersPerQuery, "netstorage.defaultMaxWorkersPerQuery()",
+		"The maximum number of CPU cores a single query can use. "+
+			"The default value should work good for most cases. "+
+			"The flag can be set to lower values for improving performance of big number of concurrently executed queries. "+
+			"The flag can be set to bigger values for improving performance of heavy queries, which scan big number of time series (>10K) and/or big number of samples (>100M). "+
+			"There is no sense in setting this flag to values bigger than the number of CPU cores available on the system")
 )
 
 // Result is a single timeseries result.

app/vmselect/netstorage/tmp_blocks_file.go

@@ -62,6 +62,9 @@ type tmpBlocksFile struct {
 	r *fs.ReaderAt
 
 	offset uint64
+
+	// err stores the first error occurred while writing the temporary blocks file.
+	err error
 }
 
 func getTmpBlocksFile() *tmpBlocksFile {
@@ -82,6 +85,7 @@ func putTmpBlocksFile(tbf *tmpBlocksFile) {
 	tbf.f = nil
 	tbf.r = nil
 	tbf.offset = 0
+	tbf.err = nil
 	tmpBlocksFilePool.Put(tbf)
 }
 
@@ -109,6 +113,10 @@ var (
 // and this must be handled.
 func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	var addr tmpBlockAddr
+	if tbf.err != nil {
+		// Do not write anything to the tbf after the first failed write
+		return addr, tbf.err
+	}
 	addr.offset = tbf.offset
 	addr.size = len(b)
 	tbf.offset += uint64(addr.size)
@@ -122,7 +130,8 @@ func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	if tbf.f == nil {
 		f, err := os.CreateTemp(tmpBlocksDir, "")
 		if err != nil {
-			return addr, err
+			tbf.err = fmt.Errorf("cannot create temporary blocks file at %q: %w", tmpBlocksDir, err)
+			return addr, tbf.err
 		}
 		tbf.f = f
 		tmpBlocksFilesCreated.Inc()
@@ -130,7 +139,9 @@ func (tbf *tmpBlocksFile) WriteBlockRefData(b []byte) (tmpBlockAddr, error) {
 	_, err := tbf.f.Write(tbf.buf)
 	tbf.buf = append(tbf.buf[:0], b...)
 	if err != nil {
-		return addr, fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
+		// The blocks buffered at tbf.buf could be partially lost, mark the tbf as unusable.
+		tbf.err = fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
+		return addr, tbf.err
 	}
 	return addr, nil
 }
@@ -141,12 +152,16 @@ func (tbf *tmpBlocksFile) Len() uint64 {
 }
 
 func (tbf *tmpBlocksFile) Finalize() error {
+	if tbf.err != nil {
+		return tbf.err
+	}
 	if tbf.f == nil {
 		return nil
 	}
 	fname := tbf.f.Name()
 	if _, err := tbf.f.Write(tbf.buf); err != nil {
-		return fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
+		tbf.err = fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
+		return tbf.err
 	}
 	tbf.buf = tbf.buf[:0]
 	r := fs.NewReaderAt(tbf.f)
@@ -166,6 +181,10 @@ func (tbf *tmpBlocksFile) Finalize() error {
 }
 
 func (tbf *tmpBlocksFile) MustReadBlockRefAt(partRef storage.PartRef, addr tmpBlockAddr) storage.BlockRef {
+	if tbf.err != nil {
+		// This should never happen, since Finalize() already returns the error for such a tbf.
+		logger.Panicf("BUG: cannot read block at %s from the temporary blocks file with the failed write: %s", addr, tbf.err)
+	}
 	var buf []byte
 	if tbf.r == nil {
 		buf = tbf.buf[addr.offset : addr.offset+uint64(addr.size)]

deployment/docker/rules/alerts-health.yml

@@ -16,6 +16,19 @@ groups:
             Job {{ $labels.job }} (instance {{ $labels.instance }}) has restarted more than twice in the last 15 minutes.
             It might be crashlooping.
 
+      - alert: UncleanShutdown
+        expr: vm_app_prev_shutdown_unclean == 1 and time() - vm_app_start_timestamp < 600
+        labels:
+          severity: warning
+        annotations:
+          summary: "{{ $labels.job }} on instance {{ $labels.instance }} started after an unclean shutdown"
+          description: |
+            The previous process run didn't shut down cleanly. Check the logs for OOM, SIGKILL,
+            a host failure, or another unexpected termination. In Kubernetes, a pod may be forcefully
+            killed with SIGKILL if the shutdown takes longer than terminationGracePeriodSeconds.
+            This alert stops firing 10 minutes after startup.
+            See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8443 for more details.
+
       - alert: ServiceDown
         expr: up{job=~".*(victoriametrics|vmselect|vminsert|vmstorage|vmagent|vmalert|vmsingle|vmalertmanager|vmauth).*"} == 0
         for: 2m

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)
+            max by (group,file) (vmalert_group_rule_results_limit * 0.9)
           )
             and on(group,file)
-          (vmalert_group_rule_results_limit > 0)
+          (max 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)
+            max by (group,file) (vmalert_group_rule_results_limit * 0.9)
           )
             and on(group,file)
-          (vmalert_group_rule_results_limit > 0)
+          (max by (group,file) (vmalert_group_rule_results_limit) > 0)
         for: 5m
         labels:
           severity: warning

docs/Makefile

@@ -90,12 +90,9 @@ endif
 	sed -i 's/\t/ /g' docs/victoriametrics/victoria_metrics_common_flags.md
 	sed -i 's/\t/ /g' docs/victoriametrics/victoria_metrics_enterprise_flags.md
 
-	# adjust flags with dynamic default values
-	# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
-	sed -i '/The maximum number of concurrent insert requests/ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/victoria_metrics_common_flags.md
-	sed -i '/The maximum number of concurrent search requests\./ s/(default [0-9]\+)/(default vmselect.getDefaultMaxConcurrentRequests())/' docs/victoriametrics/victoria_metrics_common_flags.md
-	sed -i '/The maximum number of CPU cores a single query can use\./ s/(default [0-9]\+)/(default netstorage.defaultMaxWorkersPerQuery())/' docs/victoriametrics/victoria_metrics_common_flags.md
-	sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/victoria_metrics_common_flags.md
+	# hide the machine-specific value of dynamic defaults, keeping the formula.
+	# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
+	sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/victoria_metrics_common_flags.md
 
 docs-update-vmauth-flags:
 ifndef TAG
@@ -119,9 +116,9 @@ endif
 	sed -i 's/\t/ /g' docs/victoriametrics/vmauth_common_flags.md
 	sed -i 's/\t/ /g' docs/victoriametrics/vmauth_enterprise_flags.md
 
-	# adjust flags with dynamic default values
-	# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
-	sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmauth_common_flags.md
+	# hide the machine-specific value of dynamic defaults, keeping the formula.
+	# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
+	sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmauth_common_flags.md
 
 docs-update-vmagent-flags:
 ifndef TAG
@@ -145,11 +142,9 @@ endif
 	sed -i 's/\t/ /g' docs/victoriametrics/vmagent_common_flags.md
 	sed -i 's/\t/ /g' docs/victoriametrics/vmagent_enterprise_flags.md
 
-	# adjust flags with dynamic default values
-	# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
-	sed -i '/The maximum number of concurrent insert requests/ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmagent_common_flags.md
-	sed -i '/The number of concurrent queues to each -remoteWrite.url./ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmagent_common_flags.md
-	sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmagent_common_flags.md
+	# hide the machine-specific value of dynamic defaults, keeping the formula.
+	# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
+	sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmagent_common_flags.md
 
 docs-update-vmalert-flags:
 ifndef TAG
@@ -173,10 +168,9 @@ endif
 	sed -i 's/\t/ /g' docs/victoriametrics/vmalert_common_flags.md
 	sed -i 's/\t/ /g' docs/victoriametrics/vmalert_enterprise_flags.md
 
-	# adjust flags with dynamic default values
-	# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
-	sed -i '/Defines number of writers for concurrent writing into remote write endpoint./ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmalert_common_flags.md
-	sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmalert_common_flags.md
+	# hide the machine-specific value of dynamic defaults, keeping the formula.
+	# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
+	sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmalert_common_flags.md
 
 docs-update-vmselect-flags:
 ifndef TAG

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,11 +26,14 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), `vmstorage` and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): expose the `vm_app_prev_shutdown_unclean` gauge. It is set to `1` when the previous process run didn't shut down cleanly. Added the `UncleanShutdown` [alerting rule](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-health.yml), which fires for 10 minutes after an unclean shutdown is detected. See [#8443](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8443).
 * FEATURE: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): show the selected time zone UTC offset next to the date/time controls and allow opening time zone settings from it. See [#11332](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11332).
-* FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): skip a redundant pending state when an alert can be [restored](https://docs.victoriametrics.com/victoriametrics/vmalert/#alerts-state-on-restarts) directly to firing after restart. See [#11401](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11401).
+* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/), and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): show how the default value is calculated for command-line flags which derive it from the number of available CPU cores. For example, `-maxConcurrentInserts` now prints `(default 16 = 2*cgroup.AvailableCPUs())` in `-help` output instead of `(default 16)`. Updated flags: `-search.maxConcurrentRequests`, `-search.maxWorkersPerQuery`, `-fs.maxConcurrency`, `-remoteWrite.concurrency`, `-remoteWrite.queues`. See [#9680](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680). Thanks to @Vandit1604 for contribution.
 
 * 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: `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).
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)
 

docs/victoriametrics/changelog/CHANGELOG_2025.md

@@ -115,6 +115,8 @@ Released at 2025-11-04
 
 Released at 2025-10-31
 
+**Update Note 1:** [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): reject responses with the [matrix](https://prometheus.io/docs/prometheus/latest/querying/basics/#expression-language-data-types) data type during normal rule evaluation, since vmalert expects the result to contain only a single sample or floating-point value as the rule value, not a matrix, which can contain a range of data points. Such responses could be generated by incorrect rule expressions such as `max_over_time(some_metric_filter > 90)[10m:]`, where `[10m:]` should be passed to `max_over_time` instead as `max_over_time((some_metric_filter > 90)[10m:])`.
+
 * FEATURE: `vminsert` and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): introduce new RPC protocol for insert-storage communication. See this PR [#9820](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/9820) for details.
 * FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): explicitly check response type for [range queries](https://docs.victoriametrics.com/keyConcepts.html#range-query) during [replay](https://docs.victoriametrics.com/victoriametrics/vmalert/#rules-backfilling) and return error on type mismatch. This change should reduce confusions like in [#9779](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9779).
 * FEATURE: [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): allow providing multiple filters for [remote-read migration mode](https://docs.victoriametrics.com/victoriametrics/vmctl/remoteread/) via multiple `--remote-read-filter-label` and `--remote-read-filter-label-value` flags. This is useful in order to narrow down the data being migrated by using more precise filters. See this PR [#9917](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/9917) for details.

docs/victoriametrics/integrations/grafana/_index.md

@@ -50,7 +50,7 @@ If you don't see an option to create a data source - try contacting system admin
 Create [Prometheus datasource](https://grafana.com/docs/grafana/latest/datasources/prometheus/configure/)
 in Grafana. Follow the same connection instructions as for [VictoriaMetrics datasource](#VictoriaMetrics-datasource).
 
-In the "Type and version" section set the type to "Prometheus" and the version to at least "2.24.x".
+In the "Performance" section set the Prometheus type to "Prometheus" and the Prometheus version to at least "2.24.x".
 This allows Grafana to use a more efficient API to get label values:
 
 ![Datasource](datasource-prometheus.webp)

docs/victoriametrics/vmalert.md

@@ -1127,13 +1127,18 @@ Or for all rules within the [group](#groups) {{% available_from "v1.117.0" %}}.
 Just set `debug: true` in configuration and vmalert will start printing additional log messages:
 
 ```sh
-2022-09-15T13:35:41.155Z  DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:41+02:00: query returned 0 series (elapsed: 5.896041ms, isPartial: false)
-2022-09-15T13:35:56.149Z  DEBUG datasource request: executing POST request with params "denyPartialResponse=true&query=sum%28vm_tcplistener_conns%7Binstance%3D%22localhost%3A8429%22%7D%29+by%28instance%29+%3E+0&step=15s&time=1663248945"
-2022-09-15T13:35:56.178Z  DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:56+02:00: query returned 1 series (elapsed: 28.368208ms, isPartial: false)
-2022-09-15T13:35:56.178Z  DEBUG datasource request: executing POST request with params "denyPartialResponse=true&query=sum%28vm_tcplistener_conns%7Binstance%3D%22localhost%3A8429%22%7D%29&step=15s&time=1663248945"
-2022-09-15T13:35:56.179Z  DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:56+02:00: alert 10705778000901301787 {alertgroup="TestGroup",alertname="Conns",cluster="east-1",instance="localhost:8429",replica="a"} created in state PENDING
+2026-08-20T08:21:29.464Z        info    VictoriaMetrics/app/vmalert/datasource/client.go:262    DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A20%3A00Z"
+2026-08-20T08:21:29.465Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:20:00Z: query returned 0 series (series_fetched: 0, elapsed: 1.075166ms, isPartial: false)
 ...
-2022-09-15T13:36:56.153Z  DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:36:56+02:00: alert 10705778000901301787 {alertgroup="TestGroup",alertname="Conns",cluster="east-1",instance="localhost:8429",replica="a"} PENDING => FIRING: 1m0s since becoming active at 2022-09-15 15:35:56.126006 +0200 CEST m=+39.384575417
+2026-08-20T08:22:29.466Z        info    VictoriaMetrics/app/vmalert/datasource/client.go:262    DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A21%3A00Z"
+2026-08-20T08:22:29.468Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: query returned 2 series (series_fetched: 2, elapsed: 2.055916ms, isPartial: false)
+2026-08-20T08:22:29.469Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: alert 4671711516378822929 {alertgroup="vm-health",alertname="ServiceDown",instance="victoriametrics:8428",job="victoriametrics",severity="critical"} created in state PENDING
+2026-08-20T08:22:29.469Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: alert 6230585559362831632 {alertgroup="vm-health",alertname="ServiceDown",instance="vmagent:8429",job="vmagent",severity="critical"} created in state PENDING
+...
+2026-08-20T08:23:29.463Z        info    VictoriaMetrics/app/vmalert/datasource/client.go:262    DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A22%3A00Z"
+2026-08-20T08:23:29.465Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: query returned 2 series (series_fetched: 2, elapsed: 1.391416ms, isPartial: false)
+2026-08-20T08:23:29.466Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: alert 4671711516378822929 {alertgroup="vm-health",alertname="ServiceDown",instance="victoriametrics:8428",job="victoriametrics",severity="critical"} PENDING => FIRING: 1m0s since becoming active at 2026-08-20 08:21:00 +0000 UTC
+2026-08-20T08:23:29.466Z        info    VictoriaMetrics/app/vmalert/rule/alerting.go:273        DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: alert 6230585559362831632 {alertgroup="vm-health",alertname="ServiceDown",instance="vmagent:8429",job="vmagent",severity="critical"} PENDING => FIRING: 1m0s since becoming active at 2026-08-20 08:21:00 +0000 UTC
 ```
 
 Sensitive info is stripped from the `curl` examples - see [security](#security) section for more details.

docs/victoriametrics/vmalert_common_flags.md

@@ -370,6 +370,8 @@ See the docs at https://docs.victoriametrics.com/victoriametrics/vmalert/ .
      Defines a duration for idle (keep-alive connections) to exist. Consider settings this value less to the value of "-http.idleConnTimeout". It must prevent possible "write: broken pipe" and "read: connection reset by peer" errors. (default 50s)
   -remoteWrite.maxBatchSize int
      Defines max number of timeseries to be flushed at once (default 10000)
+  -remoteWrite.maxIdleConnections int
+     Defines the number of idle (keep-alive connections) to -remoteWrite.url for the vmalert-tool debug writer, which sends every series in a separate request. Too low a value may result in a high number of sockets in TIME_WAIT state. (default 100)
   -remoteWrite.maxQueueSize int
      Defines the max number of pending datapoints to remote write endpoint (default 100000)
   -remoteWrite.oauth2.clientID string

lib/appmetrics/appmetrics.go

@@ -65,6 +65,9 @@ func writePrometheusMetrics(w io.Writer) {
 	// Export start time and uptime in seconds
 	metrics.WriteGaugeUint64(w, "vm_app_start_timestamp", uint64(startTime.Unix()))
 	metrics.WriteGaugeUint64(w, "vm_app_uptime_seconds", uint64(time.Since(startTime).Seconds()))
+	if uncleanShutdownEnabled.Load() {
+		metrics.WriteGaugeUint64(w, "vm_app_prev_shutdown_unclean", uncleanShutdown)
+	}
 
 	// Export flags as metrics.
 	isSetMap := make(map[string]bool)

lib/appmetrics/osmetrics.go

@@ -13,13 +13,13 @@ type osInfo struct {
 	release string
 }
 
-var os osInfo
+var hostOS osInfo
 var initOSOnce sync.Once
 
 func writeOSMetrics(w io.Writer) {
 	initOSOnce.Do(initOS)
 
-	if os.name != "" {
-		metrics.WriteGaugeUint64(w, fmt.Sprintf(`vm_os_info{os=%q, release=%q}`, os.name, os.release), 1)
+	if hostOS.name != "" {
+		metrics.WriteGaugeUint64(w, fmt.Sprintf(`vm_os_info{os=%q, release=%q}`, hostOS.name, hostOS.release), 1)
 	}
 }

lib/appmetrics/osmetrics_darwin.go

@@ -8,7 +8,7 @@ import (
 )
 
 func initOS() {
-	os = osInfo{name: "darwin"}
+	hostOS = osInfo{name: "darwin"}
 
 	out, err := exec.Command("sysctl", "-n", "kern.osrelease").Output()
 	if err != nil {
@@ -16,5 +16,5 @@ func initOS() {
 		return
 	}
 
-	os.release = strings.TrimSpace(string(out))
+	hostOS.release = strings.TrimSpace(string(out))
 }

lib/appmetrics/osmetrics_linux.go

@@ -7,7 +7,7 @@ import (
 )
 
 func initOS() {
-	os = osInfo{name: "linux"}
+	hostOS = osInfo{name: "linux"}
 
 	var uname syscall.Utsname
 	if err := syscall.Uname(&uname); err != nil {
@@ -22,5 +22,5 @@ func initOS() {
 		}
 		ur = append(ur, byte(v))
 	}
-	os.release = string(ur)
+	hostOS.release = string(ur)
 }

lib/appmetrics/osmetrics_windows.go

@@ -8,12 +8,12 @@ import (
 )
 
 func initOS() {
-	os = osInfo{name: "windows"}
+	hostOS = osInfo{name: "windows"}
 
 	ver := windows.RtlGetVersion()
 	if ver == nil {
 		logger.Warnf("vm_os_info metric will miss release info since windows.RtlGetVersion returned nil version")
 		return
 	}
-	os.release = fmt.Sprintf("%d.%d.%d", ver.MajorVersion, ver.MinorVersion, ver.BuildNumber)
+	hostOS.release = fmt.Sprintf("%d.%d.%d", ver.MajorVersion, ver.MinorVersion, ver.BuildNumber)
 }

lib/appmetrics/unclean_shutdown.go

@@ -0,0 +1,53 @@
+package appmetrics
+
+import (
+	"os"
+	"path/filepath"
+	"sync/atomic"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+// UncleanShutdownMarkerFilename is the marker file used to detect a previous unclean shutdown.
+const UncleanShutdownMarkerFilename = ".vm_app_running"
+
+var (
+	uncleanShutdownEnabled atomic.Bool
+	uncleanShutdown        uint64
+)
+
+// MustCreateUncleanShutdownMarker creates an UncleanShutdownMarkerFilename marker file in the given dirPath.
+// Must be called once on program startup and paired with a single MustRemoveUncleanShutdownMarker call on exit.
+//
+// If the marker file already exists on startup, it indicates a previous unclean shutdown and uncleanShutdown is set to 1.
+func MustCreateUncleanShutdownMarker(dirPath string) {
+	if !uncleanShutdownEnabled.CompareAndSwap(false, true) {
+		logger.Fatalf("BUG: unclean shutdown marker was already initialized. It could only be called once")
+	}
+	marker := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
+	f, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
+	if err == nil {
+		fs.MustClose(f)
+		return
+	}
+	if os.IsExist(err) {
+		uncleanShutdown = 1
+		logger.Warnf("Previous shutdown was unclean since file %q exists. Please check logs and investigate the reason of unclean shutdown", marker)
+		return
+	}
+	logger.Panicf("FATAL: cannot create unclean shutdown marker %q: %s", marker, err)
+}
+
+// MustRemoveUncleanShutdownMarker removes the UncleanShutdownMarkerFilename marker file created by MustCreateUncleanShutdownMarker.
+// Must be called once, as late as possible before program exit.
+func MustRemoveUncleanShutdownMarker(dirPath string) {
+	if !uncleanShutdownEnabled.Load() {
+		logger.Fatalf("BUG: unclean shutdown marker was not initialized with MustCreateUncleanShutdownMarker call")
+	}
+	marker := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
+	if err := os.Remove(marker); err != nil {
+		logger.Fatalf("FATAL: cannot remove unclean shutdown marker %q: %s", marker, err)
+	}
+	fs.MustSyncPath(dirPath)
+}

lib/appmetrics/unclean_shutdown_test.go

@@ -0,0 +1,60 @@
+package appmetrics
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+func TestUncleanShutdownLifecycle(t *testing.T) {
+
+	t.Cleanup(func() {
+		uncleanShutdownEnabled.Store(false)
+	})
+	dirPath := t.TempDir()
+	markerPath := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
+
+	// unclean logic is disabled. the unclean shutdown metric should not be exposed
+	var bb bytes.Buffer
+	writePrometheusMetrics(&bb)
+	if strings.Contains(bb.String(), "vm_app_prev_shutdown_unclean") {
+		t.Fatalf("unexpected unclean shutdown metric before starting the marker")
+	}
+
+	// clean start, the metric must report 0
+	MustCreateUncleanShutdownMarker(dirPath)
+	mustContainUncleanShutdownMetric(t, 0)
+	if _, err := os.Stat(markerPath); err != nil {
+		t.Fatalf("cannot stat the running marker after the first start: %s", err)
+	}
+	MustRemoveUncleanShutdownMarker(dirPath)
+	if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
+		t.Fatalf("unexpected running marker after a clean shutdown; got error %v; want os.ErrNotExist", err)
+	}
+	uncleanShutdownEnabled.Store(false)
+
+	// simulate prev unclean shutdown, the metric must report 1
+	if err := os.WriteFile(markerPath, nil, 0600); err != nil {
+		t.Fatalf("cannot create test marker: %s", err)
+	}
+	MustCreateUncleanShutdownMarker(dirPath)
+	mustContainUncleanShutdownMetric(t, 1)
+	MustRemoveUncleanShutdownMarker(dirPath)
+	if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
+		t.Fatalf("unexpected running marker after a clean shutdown; got error %v; want os.ErrNotExist", err)
+	}
+}
+
+func mustContainUncleanShutdownMetric(t *testing.T, value uint64) {
+	t.Helper()
+
+	var bb bytes.Buffer
+	writePrometheusMetrics(&bb)
+	want := "vm_app_prev_shutdown_unclean " + strconv.FormatUint(value, 10) + "\n"
+	if !strings.Contains(bb.String(), want) {
+		t.Fatalf("missing %q in the exported app metrics", want)
+	}
+}

lib/backup/fscommon/fscommon.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/backupnames"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
@@ -107,7 +108,7 @@ func appendFilesInternal(dst []string, d *os.File) ([]string, error) {
 }
 
 func isSpecialFile(name string) bool {
-	return name == "flock.lock" || name == backupnames.RestoreInProgressFilename || name == backupnames.RestoreMarkFileName || strings.HasSuffix(name, ".tmp")
+	return name == "flock.lock" || name == appmetrics.UncleanShutdownMarkerFilename || name == backupnames.RestoreInProgressFilename || name == backupnames.RestoreMarkFileName || strings.HasSuffix(name, ".tmp")
 }
 
 // RemoveEmptyDirs recursively removes empty directories under the given dir.

lib/flagutil/array.go

@@ -39,8 +39,29 @@ func NewArrayBool(name, description string) *ArrayBool {
 }
 
 // NewArrayInt returns new ArrayInt with the given name, defaultValue and description.
+//
+// -help shows defaultValue as a plain number. Use NewArrayIntWithDynamicDefault when
+// defaultValue is calculated at runtime.
 func NewArrayInt(name string, defaultValue int, description string) *ArrayInt {
-	description += fmt.Sprintf(" (default %d)", defaultValue)
+	return newArrayInt(name, defaultValue, strconv.Itoa(defaultValue), description)
+}
+
+// NewArrayIntWithDynamicDefault returns new ArrayInt with the given name, defaultValue and description.
+//
+// Use it instead of NewArrayInt when defaultValue is calculated at runtime.
+// See NewIntWithDynamicDefault for why such a value needs a hint.
+func NewArrayIntWithDynamicDefault(name string, defaultValue int, defaultValueHint, description string) *ArrayInt {
+	if defaultValueHint == "" {
+		panic(fmt.Sprintf("BUG: missing defaultValueHint for -%s", name))
+	}
+	return newArrayInt(name, defaultValue, fmt.Sprintf("%d = %s", defaultValue, defaultValueHint), description)
+}
+
+// newArrayInt registers an int array flag, which shows defaultValueText as its default in -help.
+//
+// Array flags keep the default in the description, since flag.Var hides an empty DefValue.
+func newArrayInt(name string, defaultValue int, defaultValueText, description string) *ArrayInt {
+	description += fmt.Sprintf(" (default %s)", defaultValueText)
 	description += "\nSupports `array` of values separated by comma or specified via multiple flags."
 	description += "\nEmpty values are set to default value."
 	a := &ArrayInt{

lib/flagutil/flag.go

@@ -7,6 +7,25 @@ import (
 	"strings"
 )
 
+// NewIntWithDynamicDefault returns a new int flag with the given name, defaultValue and description.
+//
+// Use it instead of flag.Int when defaultValue is calculated at runtime, for example
+// from the number of CPU cores. Such a value differs per machine, so -help shows both
+// the value and defaultValueHint, for example "16 = 2 * availableCPUs".
+//
+// Only -help output changes. The flag value stays defaultValue.
+func NewIntWithDynamicDefault(name string, defaultValue int, defaultValueHint, description string) *int {
+	if defaultValueHint == "" {
+		panic(fmt.Sprintf("BUG: missing defaultValueHint for -%s", name))
+	}
+	p := flag.Int(name, defaultValue, description)
+
+	// DefValue is only the text shown by -help: "default value (as text); for usage message".
+	flag.Lookup(name).DefValue = fmt.Sprintf("%d = %s", defaultValue, defaultValueHint)
+
+	return p
+}
+
 // WriteFlags writes all the explicitly set flags to w.
 func WriteFlags(w io.Writer) {
 	flag.Visit(func(f *flag.Flag) {

lib/flagutil/flag_test.go

@@ -0,0 +1,51 @@
+package flagutil
+
+import (
+	"flag"
+	"strings"
+	"testing"
+)
+
+// The flags are registered at package level, since flag registration panics when it repeats.
+var (
+	fooFlagIntDynamicDefault      = NewIntWithDynamicDefault("fooFlagIntDynamicDefault", 42, "2 * availableCPUs", "test")
+	fooFlagArrayIntDynamicDefault = NewArrayIntWithDynamicDefault("fooFlagArrayIntDynamicDefault", 42, "2 * availableCPUs", "test")
+	fooFlagArrayIntPlainDefault   = NewArrayInt("fooFlagArrayIntPlainDefault", 42, "test")
+)
+
+func TestNewIntWithDynamicDefaultSuccess(t *testing.T) {
+	// -help must show the value together with the hint.
+	f := flag.Lookup("fooFlagIntDynamicDefault")
+	if f.DefValue != "42 = 2 * availableCPUs" {
+		t.Fatalf("unexpected DefValue; got %q; want %q", f.DefValue, "42 = 2 * availableCPUs")
+	}
+
+	// the flag value must stay the calculated one.
+	if *fooFlagIntDynamicDefault != 42 {
+		t.Fatalf("unexpected flag value; got %d; want %d", *fooFlagIntDynamicDefault, 42)
+	}
+}
+
+func TestNewArrayIntWithDynamicDefaultSuccess(t *testing.T) {
+	// array flags keep the default in the description, so the hint must go there.
+	f := flag.Lookup("fooFlagArrayIntDynamicDefault")
+	if !strings.Contains(f.Usage, "(default 42 = 2 * availableCPUs)") {
+		t.Fatalf("missing the hint in the flag description; got %q", f.Usage)
+	}
+
+	// the default value must stay the calculated one.
+	if n := fooFlagArrayIntDynamicDefault.GetOptionalArg(0); n != 42 {
+		t.Fatalf("unexpected default value; got %d; want %d", n, 42)
+	}
+}
+
+func TestNewArrayIntKeepsPlainDefault(t *testing.T) {
+	// NewArrayInt must keep showing a plain number, since it shares the body with the dynamic one.
+	f := flag.Lookup("fooFlagArrayIntPlainDefault")
+	if !strings.Contains(f.Usage, "(default 42)") {
+		t.Fatalf("unexpected flag description; got %q", f.Usage)
+	}
+	if n := fooFlagArrayIntPlainDefault.GetOptionalArg(0); n != 42 {
+		t.Fatalf("unexpected default value; got %d; want %d", n, 42)
+	}
+}

lib/fs/fsutil/concurrency.go

@@ -1,14 +1,15 @@
 package fsutil
 
 import (
-	"flag"
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 )
 
-var maxConcurrency = flag.Int("fs.maxConcurrency", getDefaultConcurrency(), "The maximum number of concurrent goroutines to work with files; smaller values may help reducing Go scheduling latency "+
-	"on systems with small number of CPU cores; higher values may help reducing data ingestion latency on systems with high-latency storage such as NFS or Ceph")
+var maxConcurrency = flagutil.NewIntWithDynamicDefault("fs.maxConcurrency", getDefaultConcurrency(), "fsutil.getDefaultConcurrency()",
+	"The maximum number of concurrent goroutines to work with files; smaller values may help reducing Go scheduling latency "+
+		"on systems with small number of CPU cores; higher values may help reducing data ingestion latency on systems with high-latency storage such as NFS or Ceph")
 
 func getDefaultConcurrency() int {
 	n := min(16*cgroup.AvailableCPUs(), 256)

lib/writeconcurrencylimiter/concurrencylimiter.go

@@ -10,16 +10,18 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/metrics"
 )
 
 var (
-	maxConcurrentInserts = flag.Int("maxConcurrentInserts", 2*cgroup.AvailableCPUs(), "The maximum number of concurrent insert requests. "+
-		"Set higher value when clients send data over slow networks. "+
-		"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage. "+
-		"See also -insert.maxQueueDuration")
+	maxConcurrentInserts = flagutil.NewIntWithDynamicDefault("maxConcurrentInserts", 2*cgroup.AvailableCPUs(), "2*cgroup.AvailableCPUs()",
+		"The maximum number of concurrent insert requests. "+
+			"Set higher value when clients send data over slow networks. "+
+			"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage. "+
+			"See also -insert.maxQueueDuration")
 	maxQueueDuration = flag.Duration("insert.maxQueueDuration", time.Minute, "The maximum duration to wait in the queue when -maxConcurrentInserts "+
 		"concurrent insert requests are executed")
 )