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

app/victoria-metrics/main.go

@@ -22,7 +22,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer/push"
 )
 
 var (
@@ -104,10 +103,8 @@ func main() {
 	logger.Infof("started VictoriaMetrics in %.3f seconds", time.Since(startTime).Seconds())
 
 	pushmetrics.Init()
-	push.Init()
 	sig := procutil.WaitForSigterm()
 	logger.Infof("received signal %s", sig)
-	push.Stop()
 	pushmetrics.Stop()
 
 	stopSelfScraper()

app/vmalert/remotewrite/debug_client.go

@@ -36,13 +36,6 @@ 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

@@ -1,45 +0,0 @@
-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,7 +34,6 @@ 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")

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

app/vmselect/main.go

@@ -26,7 +26,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer/push"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/vmalertproxy"
 )
@@ -118,15 +117,8 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	// Handle non-trivial dynamic requests, which may take big amounts of time and resources.
 	startTime := time.Now()
 	defer requestDuration.UpdateDuration(startTime)
-	tracerEnabled := httputil.GetBool(r, "trace") || push.IsEnabled()
+	tracerEnabled := httputil.GetBool(r, "trace")
 	qt := querytracer.New(tracerEnabled, "%s", r.URL.Path)
-	if push.IsEnabled() {
-		defer func() {
-			if time.Since(startTime) >= push.MinTraceDuration() {
-				push.Push(qt)
-			}
-		}()
-	}
 
 	// Limit the number of concurrent queries.
 	select {

docs/victoriametrics/changelog/CHANGELOG.md

@@ -27,9 +27,9 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 ## tip
 
 * 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).
 
 * 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).
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)

docs/victoriametrics/changelog/CHANGELOG_2025.md

@@ -115,8 +115,6 @@ 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/vmalert_common_flags.md

@@ -370,8 +370,6 @@ 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/querytracer/otlp.go

@@ -1,122 +0,0 @@
-package querytracer
-
-import (
-	"crypto/rand"
-	"encoding/hex"
-	"time"
-)
-
-// OTLPTrace is an intermediate representation of a finished tracer tree
-// suitable for conversion to OTLP protobuf spans.
-type OTLPTrace struct {
-	TraceID string
-	Spans   []OTLPSpan
-}
-
-// OTLPSpan represents a single span ready for OTLP export.
-type OTLPSpan struct {
-	TraceID      string
-	SpanID       string
-	ParentSpanID string // empty for root
-	Name         string
-	StartNano    uint64
-	EndNano      uint64
-	Events       []OTLPEvent
-}
-
-// OTLPEvent is a timestamped annotation within a span (from Printf leaf nodes).
-type OTLPEvent struct {
-	TimeNano uint64
-	Name     string
-}
-
-// ToOTLPTrace converts the finished tracer tree into a flat list of OTLP spans.
-// It must be called after Done/Donef.
-func (t *Tracer) ToOTLPTrace() *OTLPTrace {
-	if t == nil {
-		return nil
-	}
-	traceID := newTraceID()
-	tr := &OTLPTrace{TraceID: traceID}
-	collectSpans(tr, t, traceID, "", t.startTime)
-	return tr
-}
-
-// collectSpans recursively walks the tracer tree and populates tr.Spans.
-// parentSpanID is empty for the root.
-// prevTime is used only when approximating timestamps for JSON-embedded spans.
-func collectSpans(tr *OTLPTrace, t *Tracer, traceID, parentSpanID string, prevTime time.Time) {
-	if t.span != nil {
-		collectJSONSpans(tr, t.span, traceID, parentSpanID, prevTime)
-		return
-	}
-
-	isLeaf := t.doneTime.Equal(t.startTime)
-	if isLeaf {
-		// Printf leaf: no span to emit here; the caller handles attaching it.
-		return
-	}
-
-	// Regular span.
-	spanID := newSpanID()
-	spanIdx := len(tr.Spans)
-	tr.Spans = append(tr.Spans, OTLPSpan{
-		TraceID:      traceID,
-		SpanID:       spanID,
-		ParentSpanID: parentSpanID,
-		Name:         t.message,
-		StartNano:    uint64(t.startTime.UnixNano()),
-		EndNano:      uint64(t.doneTime.UnixNano()),
-	})
-
-	// Process children. Leaf children become events on this span; non-leaf
-	// children recurse and produce their own spans.
-	childPrev := t.startTime
-	for _, child := range t.children {
-		if child.span == nil && child.doneTime.Equal(child.startTime) {
-			// Printf leaf: attach as span event.
-			tr.Spans[spanIdx].Events = append(tr.Spans[spanIdx].Events, OTLPEvent{
-				TimeNano: uint64(child.startTime.UnixNano()),
-				Name:     child.message,
-			})
-			continue
-		}
-		collectSpans(tr, child, traceID, spanID, childPrev)
-		if !child.doneTime.IsZero() && !child.doneTime.Equal(child.startTime) {
-			childPrev = child.doneTime
-		}
-	}
-}
-
-// collectJSONSpans converts a span tree that was deserialized via AddJSON.
-// Timestamps are approximated from prevTime since JSON spans only carry duration_msec.
-func collectJSONSpans(tr *OTLPTrace, s *span, traceID, parentSpanID string, prevTime time.Time) {
-	startNano := uint64(prevTime.UnixNano())
-	durationNano := uint64(s.DurationMsec * float64(time.Millisecond))
-	spanID := newSpanID()
-	tr.Spans = append(tr.Spans, OTLPSpan{
-		TraceID:      traceID,
-		SpanID:       spanID,
-		ParentSpanID: parentSpanID,
-		Name:         s.Message,
-		StartNano:    startNano,
-		EndNano:      startNano + durationNano,
-	})
-	childPrev := prevTime
-	for _, child := range s.Children {
-		collectJSONSpans(tr, child, traceID, spanID, childPrev)
-		childPrev = childPrev.Add(time.Duration(child.DurationMsec * float64(time.Millisecond)))
-	}
-}
-
-func newTraceID() string {
-	var b [16]byte
-	_, _ = rand.Read(b[:])
-	return hex.EncodeToString(b[:])
-}
-
-func newSpanID() string {
-	var b [8]byte
-	_, _ = rand.Read(b[:])
-	return hex.EncodeToString(b[:])
-}

lib/querytracer/push/pb.go

@@ -1,156 +0,0 @@
-package push
-
-// This file contains marshal-only protobuf types for OTLP trace export.
-// Copied and trimmed from VictoriaTraces/lib/protoparser/opentelemetry/pb/
-// Only marshaling is needed since this package only pushes traces outward.
-
-import (
-	"encoding/hex"
-
-	"github.com/VictoriaMetrics/easyproto"
-)
-
-var mp easyproto.MarshalerPool
-
-// exportTraceServiceRequest is the top-level OTLP protobuf message.
-type exportTraceServiceRequest struct {
-	ResourceSpans []*resourceSpans
-}
-
-func (r *exportTraceServiceRequest) marshalProtobuf(dst []byte) []byte {
-	m := mp.Get()
-	mm := m.MessageMarshaler()
-	for _, rs := range r.ResourceSpans {
-		rs.marshalProtobuf(mm.AppendMessage(1))
-	}
-	dst = m.Marshal(dst)
-	mp.Put(m)
-	return dst
-}
-
-// resourceSpans groups spans from a single resource (e.g. service instance).
-type resourceSpans struct {
-	Resource   resource
-	ScopeSpans []*scopeSpans
-}
-
-func (rs *resourceSpans) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	rs.Resource.marshalProtobuf(mm.AppendMessage(1))
-	for _, ss := range rs.ScopeSpans {
-		ss.marshalProtobuf(mm.AppendMessage(2))
-	}
-}
-
-// resource holds resource-level attributes (service.name, etc.).
-type resource struct {
-	Attributes []*keyValue
-}
-
-func (r *resource) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	for _, a := range r.Attributes {
-		a.marshalProtobuf(mm.AppendMessage(1))
-	}
-}
-
-// scopeSpans groups spans from a single instrumentation scope.
-type scopeSpans struct {
-	Scope instrumentationScope
-	Spans []*span
-}
-
-func (ss *scopeSpans) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	ss.Scope.marshalProtobuf(mm.AppendMessage(1))
-	for _, s := range ss.Spans {
-		s.marshalProtobuf(mm.AppendMessage(2))
-	}
-}
-
-// instrumentationScope identifies the library that produced the spans.
-type instrumentationScope struct {
-	Name    string
-	Version string
-}
-
-func (is *instrumentationScope) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	mm.AppendString(1, is.Name)
-	mm.AppendString(2, is.Version)
-}
-
-// span represents a single operation within a trace.
-type span struct {
-	// TraceID is a 32-char lowercase hex string (16 bytes).
-	TraceID string
-	// SpanID is a 16-char lowercase hex string (8 bytes).
-	SpanID string
-	// ParentSpanID is a 16-char lowercase hex string; empty for root spans.
-	ParentSpanID      string
-	Name              string
-	StartTimeUnixNano uint64
-	EndTimeUnixNano   uint64
-	Attributes        []*keyValue
-	Events            []*spanEvent
-}
-
-func (s *span) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	traceID, err := hex.DecodeString(s.TraceID)
-	if err != nil {
-		traceID = []byte(s.TraceID)
-	}
-	mm.AppendBytes(1, traceID)
-
-	spanID, err := hex.DecodeString(s.SpanID)
-	if err != nil {
-		spanID = []byte(s.SpanID)
-	}
-	mm.AppendBytes(2, spanID)
-
-	// field 3: trace_state — omitted
-
-	parentSpanID, err := hex.DecodeString(s.ParentSpanID)
-	if err != nil {
-		parentSpanID = []byte(s.ParentSpanID)
-	}
-	mm.AppendBytes(4, parentSpanID)
-
-	mm.AppendString(5, s.Name)
-	// field 6: kind — omitted (INTERNAL=1 is default)
-	mm.AppendFixed64(7, s.StartTimeUnixNano)
-	mm.AppendFixed64(8, s.EndTimeUnixNano)
-	for _, a := range s.Attributes {
-		a.marshalProtobuf(mm.AppendMessage(9))
-	}
-	for _, e := range s.Events {
-		e.marshalProtobuf(mm.AppendMessage(11))
-	}
-}
-
-// spanEvent is a time-stamped annotation within a span.
-type spanEvent struct {
-	TimeUnixNano uint64
-	Name         string
-}
-
-func (se *spanEvent) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	mm.AppendFixed64(1, se.TimeUnixNano)
-	mm.AppendString(2, se.Name)
-}
-
-// keyValue is an OTLP attribute key-value pair.
-type keyValue struct {
-	Key   string
-	Value anyValue
-}
-
-func (kv *keyValue) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	mm.AppendString(1, kv.Key)
-	kv.Value.marshalProtobuf(mm.AppendMessage(2))
-}
-
-// anyValue holds a single string attribute value (sufficient for our use case).
-type anyValue struct {
-	StringValue string
-}
-
-func (av *anyValue) marshalProtobuf(mm *easyproto.MessageMarshaler) {
-	mm.AppendString(1, av.StringValue)
-}

lib/querytracer/push/push.go

@@ -1,233 +0,0 @@
-// Package push implements background exporting of VictoriaMetrics query traces
-// to VictoriaTraces in OTLP protobuf format over HTTP.
-package push
-
-import (
-	"bytes"
-	"compress/gzip"
-	"flag"
-	"fmt"
-	"net/http"
-	"sync"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
-	"github.com/VictoriaMetrics/metrics"
-)
-
-var (
-	pushURL = flag.String("search.traceExportURL", "", "If set, query traces are exported to this URL in OTLP protobuf format. "+
-		"For example, -search.traceExportURL=http://victoria-traces:4318/insert/opentelemetry/v1/traces . "+
-		"See https://docs.victoriametrics.com/victoriametrics/query-tracing/")
-	minTraceDuration = flag.Duration("search.traceExportMinDuration", 0, "Minimum query duration for exporting traces via -search.traceExportURL. "+
-		"Traces for faster queries are dropped. 0 means all traces are exported.")
-)
-
-var (
-	pushesTotal  = metrics.NewCounter(`vm_trace_export_pushes_total`)
-	errorsTotal  = metrics.NewCounter(`vm_trace_export_errors_total`)
-	droppedTotal = metrics.NewCounter(`vm_trace_export_dropped_total`)
-)
-
-// queueCap is the maximum number of pending traces waiting to be exported.
-const queueCap = 1000
-
-var (
-	queue chan *querytracer.OTLPTrace
-
-	stopCh chan struct{}
-	wg     sync.WaitGroup
-)
-
-// IsEnabled returns true when trace export is configured.
-func IsEnabled() bool {
-	return *pushURL != ""
-}
-
-// MinTraceDuration returns the configured minimum trace duration threshold.
-func MinTraceDuration() time.Duration {
-	return *minTraceDuration
-}
-
-// Init starts the background export goroutine. Must be called after flag.Parse and logger.Init.
-func Init() {
-	if !IsEnabled() {
-		return
-	}
-	queue = make(chan *querytracer.OTLPTrace, queueCap)
-	stopCh = make(chan struct{})
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		runExporter()
-	}()
-	logger.Infof("started query trace exporter to %s", *pushURL)
-}
-
-// Stop drains pending traces and stops the background goroutine.
-// Must be called during graceful shutdown before the process exits.
-func Stop() {
-	if !IsEnabled() {
-		return
-	}
-	close(stopCh)
-	wg.Wait()
-}
-
-// Push enqueues t for async export. t must be a finished tracer.
-// Traces are silently dropped if the queue is full.
-func Push(t *querytracer.Tracer) {
-	if !t.Enabled() {
-		return
-	}
-	tr := t.ToOTLPTrace()
-	if tr == nil {
-		return
-	}
-	select {
-	case queue <- tr:
-	default:
-		droppedTotal.Inc()
-	}
-}
-
-// runExporter reads from queue and sends batches to the configured endpoint.
-func runExporter() {
-	ticker := time.NewTicker(time.Second)
-	defer ticker.Stop()
-
-	var batch []*querytracer.OTLPTrace
-	for {
-		select {
-		case tr := <-queue:
-			batch = append(batch, tr)
-			if len(batch) >= 100 {
-				flushBatch(batch)
-				batch = batch[:0]
-			}
-		case <-ticker.C:
-			if len(batch) > 0 {
-				flushBatch(batch)
-				batch = batch[:0]
-			}
-		case <-stopCh:
-			// Drain remaining items.
-			for {
-				select {
-				case tr := <-queue:
-					batch = append(batch, tr)
-				default:
-					if len(batch) > 0 {
-						flushBatch(batch)
-					}
-					return
-				}
-			}
-		}
-	}
-}
-
-// flushBatch serializes and POSTs a batch of traces.
-func flushBatch(batch []*querytracer.OTLPTrace) {
-	data, err := marshalBatch(batch)
-	if err != nil {
-		logger.Errorf("cannot marshal trace batch: %s", err)
-		errorsTotal.Inc()
-		return
-	}
-	if err := postData(data); err != nil {
-		logger.Warnf("cannot export query traces to %s: %s", *pushURL, err)
-		errorsTotal.Inc()
-		return
-	}
-	pushesTotal.Add(len(batch))
-}
-
-// marshalBatch packs all traces into a single ExportTraceServiceRequest and gzip-compresses it.
-func marshalBatch(batch []*querytracer.OTLPTrace) ([]byte, error) {
-	svcVersion := buildinfo.Version
-
-	// Build resource attributes once (same for all spans in this process).
-	resAttrs := []*keyValue{
-		{Key: "service.name", Value: anyValue{StringValue: "victoriametrics"}},
-		{Key: "service.version", Value: anyValue{StringValue: svcVersion}},
-	}
-
-	var pbSpans []*span
-	for _, tr := range batch {
-		for i := range tr.Spans {
-			s := &tr.Spans[i]
-			pbSpan := &span{
-				TraceID:           s.TraceID,
-				SpanID:            s.SpanID,
-				ParentSpanID:      s.ParentSpanID,
-				Name:              s.Name,
-				StartTimeUnixNano: s.StartNano,
-				EndTimeUnixNano:   s.EndNano,
-			}
-			for _, ev := range s.Events {
-				pbSpan.Events = append(pbSpan.Events, &spanEvent{
-					TimeUnixNano: ev.TimeNano,
-					Name:         ev.Name,
-				})
-			}
-			pbSpans = append(pbSpans, pbSpan)
-		}
-	}
-
-	req := &exportTraceServiceRequest{
-		ResourceSpans: []*resourceSpans{
-			{
-				Resource: resource{Attributes: resAttrs},
-				ScopeSpans: []*scopeSpans{
-					{
-						Scope: instrumentationScope{
-							Name:    "victoriametrics/querytracer",
-							Version: svcVersion,
-						},
-						Spans: pbSpans,
-					},
-				},
-			},
-		},
-	}
-
-	raw := req.marshalProtobuf(nil)
-
-	var buf bytes.Buffer
-	gz := gzip.NewWriter(&buf)
-	if _, err := gz.Write(raw); err != nil {
-		return nil, fmt.Errorf("cannot gzip trace data: %w", err)
-	}
-	if err := gz.Close(); err != nil {
-		return nil, fmt.Errorf("cannot close gzip writer: %w", err)
-	}
-	return buf.Bytes(), nil
-}
-
-var httpClient = &http.Client{
-	Timeout: 10 * time.Second,
-}
-
-// postData sends gzip-compressed protobuf data to the configured endpoint.
-func postData(data []byte) error {
-	req, err := http.NewRequest(http.MethodPost, *pushURL, bytes.NewReader(data))
-	if err != nil {
-		return fmt.Errorf("cannot create HTTP request: %w", err)
-	}
-	req.Header.Set("Content-Type", "application/x-protobuf")
-	req.Header.Set("Content-Encoding", "gzip")
-
-	resp, err := httpClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("HTTP request failed: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode/100 != 2 {
-		return fmt.Errorf("unexpected HTTP response status %d", resp.StatusCode)
-	}
-	return nil
-}