vmalert: skip a redundant pending state when an alert can be restored to firing

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) ([]prompb.TimeSeries, error) {
+func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := ar.q.Query(ctx, ar.Expr, ts)
 	curState := StateEntry{
@@ -546,6 +546,15 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 		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 {
@@ -799,7 +808,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, lookback time.Duration) error {
+func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts time.Time) error {
 	if ar.For < 1 {
 		return nil
 	}
@@ -825,11 +834,9 @@ 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(lookback.Seconds()))
+		alertForStateMetricName, nameStr, labelsFilter, int(remoteReadLookBack.Seconds()))
 
-	// 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))
+	res, _, err := q.Query(ctx, expr, ts)
 	if err != nil {
 		return fmt.Errorf("failed to execute restore query %q: %w ", expr, err)
 	}
@@ -839,9 +846,6 @@ 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)
+		_, err := ar.exec(context.TODO(), ts1, 0, nil)
 		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)
+		_, err = ar.exec(context.TODO(), ts2, 0, nil)
 		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)
+			tss, err := rule.exec(context.TODO(), ts, 0, nil)
 			if err != nil {
 				t.Fatalf("unexpected error: %s", err)
 			}
@@ -824,10 +824,11 @@ func TestAlertingRuleExecRange(t *testing.T) {
 func TestGroup_Restore(t *testing.T) {
 	defaultTS := time.Now()
 	fqr := &datasource.FakeQuerierWithRegistry{}
-	fn := func(rules []config.Rule, expAlerts map[uint64]*notifier.Alert) {
+	f := func(rules []config.Rule, expAlerts map[uint64]*notifier.Alert, expNotificationNum int) {
 		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{}
@@ -857,8 +858,8 @@ func TestGroup_Restore(t *testing.T) {
 			if !ok {
 				t.Fatalf("expected to have key %d", key)
 			}
-			if got.State != notifier.StatePending {
-				t.Fatalf("expected state %d; got %d", notifier.StatePending, got.State)
+			if got.State != exp.State {
+				t.Fatalf("expected state %d; got %d", exp.State, got.State)
 			}
 			if got.ActiveAt != exp.ActiveAt {
 				t.Fatalf("expected ActiveAt %v; got %v", exp.ActiveAt, got.ActiveAt)
@@ -867,6 +868,9 @@ 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 {
@@ -878,28 +882,30 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert, no previous state
 	fqr.Set("foo", metricWithValueAndLabels(t, 0, "__name__", "foo"))
-	fn(
+	f(
 		[]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))
-	fn(
+	f(
 		[]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)
@@ -909,7 +915,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"))
-	fn(
+	f(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
 		},
@@ -917,12 +923,14 @@ 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)
@@ -930,7 +938,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))
-	fn(
+	f(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
 			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
@@ -939,12 +947,14 @@ 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)
@@ -954,63 +964,68 @@ func TestGroup_Restore(t *testing.T) {
 		stateMetric("foo", ts))
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
 		stateMetric("bar", ts))
-	fn(
+	f(
 		[]config.Rule{
 			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
-			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
+			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Hour)},
 		},
 		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))
-	fn(
+	f(
 		[]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"))
-	fn(
+	f(
 		[]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"))
-	fn(
+	f(
 		[]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)
@@ -1020,18 +1035,20 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set("foo",
 		metricWithValueAndLabels(t, 0, "__name__", "foo", "env", "dev"),
 		metricWithValueAndLabels(t, 0, "__name__", "foo", "env", "prod"))
-	fn(
+	f(
 		[]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) {
@@ -1044,14 +1061,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)
+	_, err := ar.exec(context.TODO(), time.Now(), 0, nil)
 	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)
+	_, err = ar.exec(context.TODO(), time.Now(), 0, nil)
 	if !errors.Is(err, errDuplicate) {
 		t.Fatalf("expected to have %s error; got %s", errDuplicate, err)
 	}
@@ -1060,7 +1077,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)
+	_, err = ar.exec(context.TODO(), time.Now(), 0, nil)
 	if err == nil {
 		t.Fatalf("expected to get err; got nil")
 	}
@@ -1083,7 +1100,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)
+		_, err := ar.exec(context.TODO(), timestamp, limit, nil)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -1111,7 +1128,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)
+		_, err := ar.exec(context.TODO(), timestamp, limit, nil)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -1140,7 +1157,7 @@ func TestAlertingRule_Template(t *testing.T) {
 		fq.SetPartialResponse(isResponsePartial)
 
 		ts := time.Unix(3600, 0)
-		if _, err := rule.exec(context.TODO(), ts, 0); err != nil {
+		if _, err := rule.exec(context.TODO(), ts, 0, nil); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		for hash, expAlert := range alertsExpected {
@@ -1439,7 +1456,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)
+	_, err := ar.exec(context.TODO(), ts, 0, nil)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
@@ -1471,7 +1488,7 @@ func TestAlertingRule_QueryTemplateInLabels(t *testing.T) {
 	fq.Add(metricWithValueAndLabels(t, 1, "device", "sda1"))
 
 	ts := time.Now()
-	_, err := ar.exec(context.TODO(), ts, 0)
+	_, err := ar.exec(context.TODO(), ts, 0, nil)
 	if err != nil {
 		t.Fatalf("unexpected error with query template in labels: %s", err)
 	}

app/vmalert/rule/group.go

@@ -222,29 +222,6 @@ 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
@@ -395,7 +372,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 
 	g.infof("started")
 
-	eval := func(ctx context.Context, ts time.Time) time.Time {
+	eval := func(ctx context.Context, ts time.Time, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) {
 		g.metrics.iterationTotal.Inc()
 
 		start := time.Now()
@@ -405,13 +382,13 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 			g.mu.Lock()
 			g.LastEvaluation = start
 			g.mu.Unlock()
-			return ts
+			return
 		}
 
 		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)
+		errs := e.execConcurrently(ctx, g.Rules, ts, g.Concurrency, resolveDuration, g.Limit, getRemoteReadQuerier)
 		for err := range errs {
 			if err != nil {
 				logger.Errorf("group %q (file=%q): %s", g.Name, g.File, err)
@@ -421,7 +398,6 @@ 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)
@@ -436,16 +412,19 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 	t := time.NewTicker(g.Interval)
 	defer t.Stop()
 
-	realEvalTS := eval(evalCtx, evalTS)
-
-	// restore the rules state after the first evaluation
-	// so only active alerts can be restored.
+	var getRemoteReadQuerier func(enableDebug bool) datasource.Querier
 	if rr != nil {
-		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)
+		getRemoteReadQuerier = func(enableDebug bool) datasource.Querier {
+			return rr.BuildWithParams(datasource.QuerierParams{
+				EvaluationInterval: g.Interval,
+				QueryParams:        g.Params,
+				Headers:            g.Headers,
+				Debug:              enableDebug,
+			})
 		}
 	}
+	// pass getRemoteReadQuerier to the first evaluation, so it can be used for restoring alert states
+	eval(evalCtx, evalTS, getRemoteReadQuerier)
 
 	for {
 		select {
@@ -496,7 +475,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 				g.metrics.iterationMissed.Inc()
 			}
 
-			eval(evalCtx, evalTS)
+			eval(evalCtx, evalTS, nil)
 		}
 	}
 }
@@ -667,7 +646,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)
+	return e.execConcurrently(ctx, g.Rules, evalTS, g.Concurrency, resolveDuration, g.Limit, nil)
 }
 
 type rangeIterator struct {
@@ -741,12 +720,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) chan error {
+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 {
 	res := make(chan error, len(rules))
 	if concurrency == 1 {
 		// fast path
 		for _, rule := range rules {
-			res <- e.exec(ctx, rule, ts, resolveDuration, limit)
+			res <- e.exec(ctx, rule, ts, resolveDuration, limit, getRemoteReadQuerier)
 		}
 		close(res)
 		return res
@@ -759,7 +738,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)
+				res <- e.exec(ctx, rule, ts, resolveDuration, limit, getRemoteReadQuerier)
 				<-sem
 			})
 		}
@@ -776,10 +755,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) error {
+func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDuration time.Duration, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) error {
 	execTotal.Inc()
 
-	tss, err := r.exec(ctx, ts, limit)
+	tss, err := r.exec(ctx, ts, limit, getRemoteReadQuerier)
 	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)
+		_ = e.exec(ctx, r, time.Now(), 0, 10, nil)
 	}()
 
 	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)
+	err := e.exec(context.Background(), r, time.Now(), 0, 10, nil)
 	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) ([]prompb.TimeSeries, error) {
+func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int, _ func(enableDebug bool) datasource.Querier) ([]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)
+			tss, err := rule.exec(context.TODO(), ts, 0, nil)
 			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)
+		_, err := rule.exec(context.TODO(), time.Now(), limit, nil)
 		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)
+		_, err := rule.exec(context.TODO(), time.Now(), limit, nil)
 		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)
+	_, err := rr.exec(context.TODO(), time.Now(), 0, nil)
 	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)
+	_, err = rr.exec(context.TODO(), time.Now(), 0, nil)
 	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)
+	got, err := rule.exec(context.TODO(), ts, 0, nil)
 	want := []prompb.TimeSeries{
 		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompb.Label{
 			{

app/vmalert/rule/rule.go

@@ -10,6 +10,7 @@ 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"
@@ -27,7 +28,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) ([]prompb.TimeSeries, error)
+	exec(ctx context.Context, ts time.Time, limit int, getRemoteReadQuerier func(enableDebug bool) datasource.Querier) ([]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

docs/victoriametrics/README.md

@@ -1266,96 +1266,47 @@ See also [resource usage limits at VictoriaMetrics cluster](https://docs.victori
 
 ## High availability
 
-VictoriaMetrics supports high availability for both writes and reads by combining replication with multiple instances.
+The general approach for achieving high availability is the following:
 
-You can achieve **high availability for writes** using replication:
+* To run two identically configured VictoriaMetrics instances in distinct datacenters (availability zones);
+* To store the collected data simultaneously into these instances via [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) or Prometheus.
+* To query the first VictoriaMetrics instance and to fail over to the second instance when the first instance becomes temporarily unavailable.
+  This can be done via [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/) according to [these docs](https://docs.victoriametrics.com/victoriametrics/vmauth/#high-availability).
 
-* Run two or more identically configured VictoriaMetrics instances in distinct datacenters (availability zones);
-* Replicate collected metrics simultaneously into all these instances via one or more [vmagents](https://docs.victoriametrics.com/victoriametrics/vmagent/).
+Such a setup guarantees that the collected data isn't lost when one of VictoriaMetrics instance becomes unavailable.
+The collected data continues to be written to the available VictoriaMetrics instance, so it should be available for querying.
+Both [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and Prometheus buffer the collected data locally if they cannot send it
+to the configured remote storage. So the collected data will be written to the temporarily unavailable VictoriaMetrics instance
+after it becomes available.
+
+If you use [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) for storing the data into VictoriaMetrics,
+then it can be configured with multiple `-remoteWrite.url` command-line flags, where every flag points to the VictoriaMetrics
+instance in a particular availability zone, in order to replicate the collected data to all the VictoriaMetrics instances.
+For example, the following command instructs `vmagent` to replicate data to `vm-az1` and `vm-az2` instances of VictoriaMetrics:
 
-In this setup, configure vmagent [to replicate data](https://docs.victoriametrics.com/victoriametrics/vmagent/#replication-and-high-availability)
-to each remote destination:
 ```sh
 /path/to/vmagent \
-  -remoteWrite.url=https://victoriametrics-1:8428/api/v1/write \
-  -remoteWrite.url=https://victoriametrics-2:8428/api/v1/write
+  -remoteWrite.url=http://<vm-az1>:8428/api/v1/write \
+  -remoteWrite.url=http://<vm-az2>:8428/api/v1/write
 ```
 
-Each `--remoteWrite.url` creates its own replication queue. The queue temporarily stores data on disk while a remote destination is unavailable.
-See more about [on-disk persistence in vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/#on-disk-persistence).
+If you use Prometheus for collecting and writing the data to VictoriaMetrics,
+then the following [`remote_write`](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#remote_write) section
+in Prometheus config can be used for replicating the collected data to `vm-az1` and `vm-az2` VictoriaMetrics instances:
 
-When the remote destination becomes available, vmagent drains the queue and restores data consistency across destinations.
-
-> The max size of the on-disk queue can be increased by [horizontally sharding vmagents](https://docs.victoriametrics.com/victoriametrics/vmagent/#scraping-big-number-of-targets).
-> To achieve high availability for vmagent itself, run multiple identically configured vmagent replicas.
-> In this case, the load on the remote destinations will increase proportionally to the number of vmagent replicas. The duplicated data in remote destinations
-> has to be [deduplicated](https://docs.victoriametrics.com/victoriametrics/#deduplication) on the VictoriaMetrics side.
-
-You can achieve **high availability for reads** by choosing one of the following options:
-
-- Load balancer: Use a load balancer to ensure read operations are always routed to an available VictoriaMetrics instance.
-- Top-level vmselect: Use vmselect to query all available VictoriaMetrics instances and merge the results
-
-### Load balancer
-
-In this mode, we use a load balancer to query the main VictoriaMetrics instance and fail over to a secondary instance if the first one becomes temporarily unavailable.
-
-This can be done using [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/) configured in [high-availability mode](https://docs.victoriametrics.com/victoriametrics/vmauth/#high-availability).
-
-```mermaid
-flowchart LR
-    Client["Query Client<br/>Grafana/vmalert"]
-
-    VMAUTH["vmauth<br/>Load Balancer / Failover"]
-
-    VM1["VictoriaMetrics-1<br/>Primary read target"]
-    VM2["VictoriaMetrics-2<br/>Failover read target"]
-
-    Client -->|"Read query"| VMAUTH
-
-    VMAUTH -->|"1. Send queries"| VM1
-    VMAUTH -.->|"2. Fail over if VM1<br/>is unavailable"| VM2
+```yaml
+remote_write:
+  - url: http://<vm-az1>:8428/api/v1/write
+  - url: http://<vm-az2>:8428/api/v1/write
 ```
 
-This is the most cost-efficient option because only one VictoriaMetrics instance is queried at a time.
+It is recommended to use [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) instead of Prometheus for highly loaded setups,
+since it uses lower amounts of RAM, CPU and network bandwidth than Prometheus.
 
-The downside is that when one instance goes down and then comes back up, the load balancer may immediately start sending read queries to the recovering instance, even though it hasn't caught up yet. During this short recovery window:
+If you use identically configured [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) instances for collecting the same data
+and sending it to VictoriaMetrics, then do not forget enabling [deduplication](#deduplication) at VictoriaMetrics side.
 
-- The recovering VictoriaMetrics instance has not yet ingested all the data that was queued while it was down.
-- vmagent nodes may still be draining their backlogged queues into that instance
-- Read queries routed to the recovering instance can return incomplete results, because some recent data may still be in transit or not yet visible.
-
-### Top-level vmselect
-
-In this option, we use a top-level [vmselect](https://docs.victoriametrics.com/victoriametrics/vmselect/) to query all remote destinations simultaneously and merge the results.
-
-This option is only possible if VictoriaMetrics single-node instances are configured with the `-vmselectAddr` flag. See more details in the [VictoriaMetrics multi-tenancy section](https://docs.victoriametrics.com/victoriametrics/#multi-tenancy).
-
-```mermaid
-flowchart LR
-    Client["Query Client<br/>Grafana / vmalert"]
-
-    VMSELECT["vmselect<br/>Query all destinations<br/>
-     <code><pre>-dedup.minScrapeInterval=1ms<br/>-replicationFactor=2</pre></code>"]
-
-    VM1["VictoriaMetrics-1<br/>Single-node<br/><code>-vmselectAddr=:8401</code>"]
-    VM2["VictoriaMetrics-2<br/>Single-node<br/><code>-vmselectAddr=:8401</code>"]
-
-    Client -->|"Read query"| VMSELECT
-
-    VMSELECT --> VM1
-    VMSELECT --> VM2
-
-    VMSELECT -->|"Merged and deduplicated results"| Client
-```
-
-This option requires more resources because it queries all remote destinations simultaneously and merges their responses before returning the final result.
-
-The benefit is that it can handle data gaps across destinations by merging responses from all VictoriaMetrics instances. Thus, a single recovering instance is less likely to cause incomplete results, unlike in the load balancer case.
-
-Since vmselect queries identical data across multiple VictoriaMetrics instances, configure it with `-dedup.minScrapeInterval=1ms` to remove duplicate samples during merging. Also set `-replicationFactor=N` on vmselect, where `N` equals the number of remote storage destinations, so that queries can tolerate the unavailability of up to `N-1` destinations.
-
-See [VMDistributed](https://docs.victoriametrics.com/operator/resources/vmdistributed/) Kubernetes operator resource for an example of configuring vmselect with deduplication and replication.
+See [VMDistributed](https://docs.victoriametrics.com/operator/resources/vmdistributed/) Kubernetes operator resource for an example.
 
 ## Deduplication
 

docs/victoriametrics/changelog/CHANGELOG.md

@@ -35,6 +35,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * FEATURE: [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/): extend `-replay.continueWithExecutionErr` to also handle the `400 Bad Request` response code, since it is used for Prometheus querying API requests when request parameters are missing or incorrect. See [#11352](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11352).
 * FEATURE: `vmselect` and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/), and `vmagent`: set default value of `-enableMultitenancyViaHeaders` to `true`. This change enables support of [multitenancy via headers for cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#multitenancy-via-headers) and [for vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/#multitenancy-via-headers) by default, aligning VictoriaMetrics multitenancy behavior with [multitenancy in VictoriaLogs](https://docs.victoriametrics.com/victorialogs/#multitenancy). See related ticket [#11365](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11365).
 * FEATURE: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): add an option to customize the favicon color. This makes it easier to distinguish between different installations opened in multiple browser tabs. See [#11329](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11329).
+* 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).
 
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): avoid suggesting the unrelated `-enableTCP6` command-line flag when scraping a target over a Unix domain socket fails. See [#11320](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11320). Thanks to @lwmacct for contribution.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): skip labels with empty name at [/api/v1/import](https://docs.victoriametrics.com/victoriametrics/#how-to-import-data-in-json-line-format). Previously such a label replaced the metric name, so a series sent with `"metric":{"__name__":"foo","":"bar"}` was stored under the name `bar`. Other ingestion protocols already skip such labels. See [#4962](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4962). Thanks to @Vandit1604 for contribution.