vmalert: skip redundant pending state during restore

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, rr datasource.Querier) ([]prompb.TimeSeries, error) {
 	start := time.Now()
 	res, req, err := ar.q.Query(ctx, ar.Expr, ts)
 	curState := StateEntry{
@@ -541,6 +541,14 @@ 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")
 	}
+	if rr != nil {
+		// try to restore alerts state from remoteRead
+		// do not break the evaluation if restore request failed
+		err := ar.restore(ctx, rr, ts)
+		if err != nil {
+			logger.Errorf("error while restoring ruleState for group %q rule %q: %s", ar.GroupName, ar.Name, err)
+		}
+	}
 	var numActivePending int
 	var tss []prompb.TimeSeries
 	for h, a := range ar.alerts {
@@ -792,7 +800,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, rr datasource.Querier, ts time.Time) error {
 	if ar.For < 1 {
 		return nil
 	}
@@ -818,11 +826,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(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 := rr.Query(ctx, expr, ts.Add(-1*time.Second))
 	if err != nil {
 		return fmt.Errorf("failed to execute restore query %q: %w ", expr, err)
 	}
@@ -832,8 +840,8 @@ func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts ti
 		return nil
 	}
 
-	ar.alertsMu.Lock()
-	defer ar.alertsMu.Unlock()
+	// ar.alertsMu.Lock()
+	// defer ar.alertsMu.Unlock()
 
 	for _, series := range res.Data {
 		series.DelLabel("__name__")

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)
 			}
@@ -819,10 +819,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{}
@@ -852,8 +853,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)
@@ -862,6 +863,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 {
@@ -873,28 +877,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)
@@ -904,7 +910,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)},
 		},
@@ -912,12 +918,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)
@@ -925,7 +933,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)},
@@ -934,12 +942,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)
@@ -949,63 +959,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)
@@ -1015,18 +1030,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) {
@@ -1039,14 +1056,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)
 	}
@@ -1055,7 +1072,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")
 	}
@@ -1078,7 +1095,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")
 		}
@@ -1106,7 +1123,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)
 		}
@@ -1134,7 +1151,7 @@ func TestAlertingRule_Template(t *testing.T) {
 		fq.Add(metrics...)
 		fq.SetPartialResponse(isResponsePartial)
 
-		if _, err := rule.exec(context.TODO(), time.Now(), 0); err != nil {
+		if _, err := rule.exec(context.TODO(), time.Now(), 0, nil); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		for hash, expAlert := range alertsExpected {
@@ -1429,7 +1446,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)
 	}
@@ -1461,7 +1478,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

@@ -223,27 +223,27 @@ func (g *Group) CreateID() uint64 {
 }
 
 // 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
-}
+// 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
@@ -393,7 +393,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, rr datasource.Querier) {
 		g.metrics.iterationTotal.Inc()
 
 		start := time.Now()
@@ -403,13 +403,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, rr)
 		for err := range errs {
 			if err != nil {
 				logger.Errorf("group %q: %s", g.Name, err)
@@ -419,7 +419,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)
@@ -434,16 +433,25 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 	t := time.NewTicker(g.Interval)
 	defer t.Stop()
 
-	realEvalTS := eval(evalCtx, evalTS)
+	var remoteReadQuerier datasource.Querier
+	if rr != nil {
+		remoteReadQuerier = rr.BuildWithParams(datasource.QuerierParams{
+			EvaluationInterval: g.Interval,
+			QueryParams:        g.Params,
+			Headers:            g.Headers,
+			Debug:              g.Debug,
+		})
+	}
+	eval(evalCtx, evalTS, remoteReadQuerier)
 
 	// restore the rules state after the first evaluation
 	// so only active alerts can be restored.
-	if rr != nil {
-		err := g.restore(ctx, rr, realEvalTS, *remoteReadLookBack)
-		if err != nil {
-			logger.Errorf("error while restoring ruleState for group %q: %s", g.Name, err)
-		}
-	}
+	// if rr != nil {
+	// 	err := g.restore(ctx, rr, realEvalTS, *remoteReadLookBack)
+	// 	if err != nil {
+	// 		logger.Errorf("error while restoring ruleState for group %q: %s", g.Name, err)
+	// 	}
+	// }
 
 	for {
 		select {
@@ -494,7 +502,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 				g.metrics.iterationMissed.Inc()
 			}
 
-			eval(evalCtx, evalTS)
+			eval(evalCtx, evalTS, nil)
 		}
 	}
 }
@@ -665,7 +673,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 {
@@ -739,12 +747,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, rr 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, rr)
 		}
 		close(res)
 		return res
@@ -757,7 +765,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, rr)
 				<-sem
 			})
 		}
@@ -774,10 +782,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, rr datasource.Querier) error {
 	execTotal.Inc()
 
-	tss, err := r.exec(ctx, ts, limit)
+	tss, err := r.exec(ctx, ts, limit, rr)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
 			// the context can be cancelled on graceful shutdown

app/vmalert/rule/group_test.go

@@ -484,7 +484,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()
@@ -516,7 +516,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, _ 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

@@ -9,6 +9,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/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
@@ -25,7 +26,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, rr 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