app/vmselect: export traces to VictoriaTraces

README.md

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

app/victoria-metrics/main.go

@@ -22,6 +22,7 @@ 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 (
@@ -103,8 +104,10 @@ 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/vmctl/main.go

@@ -421,11 +421,6 @@ func main() {
 
 					disableKeepAlive := c.Bool(vmNativeDisableHTTPKeepAlive)
 
-					cc := c.Int(vmConcurrency)
-					if cc <= 0 {
-						cc = 1
-					}
-
 					var srcExtraLabels []string
 					srcAddr := strings.Trim(c.String(vmNativeSrcAddr), "/")
 					srcAuthConfig, err := auth.Generate(
@@ -451,8 +446,6 @@ func main() {
 					trSrc := httputil.NewTransport(false, "vmctl_src")
 					trSrc.DisableKeepAlives = disableKeepAlive
 					trSrc.TLSClientConfig = srcTC
-					// Keep an idle connection per worker to reduce connections churn.
-					trSrc.MaxIdleConnsPerHost = cc
 
 					srcHTTPClient := &http.Client{
 						Transport: trSrc,
@@ -483,8 +476,6 @@ func main() {
 					trDst := httputil.NewTransport(false, "vmctl_dst")
 					trDst.DisableKeepAlives = disableKeepAlive
 					trDst.TLSClientConfig = dstTC
-					// Keep an idle connection per worker to reduce connections churn.
-					trDst.MaxIdleConnsPerHost = cc
 
 					dstHTTPClient := &http.Client{
 						Transport: trDst,
@@ -513,7 +504,7 @@ func main() {
 							HTTPClient:  dstHTTPClient,
 						},
 						backoff:                  bf,
-						cc:                       cc,
+						cc:                       c.Int(vmConcurrency),
 						disablePerMetricRequests: c.Bool(vmNativeDisablePerMetricMigration),
 						isNative:                 !c.Bool(vmNativeDisableBinaryProtocol),
 					}

app/vmselect/main.go

@@ -26,6 +26,7 @@ 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"
 )
@@ -117,8 +118,15 @@ 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")
+	tracerEnabled := httputil.GetBool(r, "trace") || push.IsEnabled()
 	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

@@ -31,7 +31,6 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix infinite loop in the OpenTelemetry Firehose ingestion endpoint (`/opentelemetry/api/v1/push`) when receiving a malformed record with an incomplete varint in the `data` field. Previously this caused the goroutine to spin forever, permanently consuming CPU until the process was restarted.
 * BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/): reuse connections to `-remoteWrite.url` when writing the results of recording rules and alerts. Previously every series was sent over a new connection, which left a lot of sockets in `TIME_WAIT` state and could exhaust the ephemeral port range. The number of idle connections can be tuned via the new `-remoteWrite.maxIdleConnections` command-line flag. Thanks @evkuzin for contribution.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): prevent process crash in `sort_by_label_numeric()` and `sort_by_label_numeric_desc()` when a label value contains a number with 309 or more digits. See [#11423](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11423).
-* BUGFIX: [vmctl](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): reuse connections in [vm-native mode](https://docs.victoriametrics.com/victoriametrics/vmctl/#migrating-data-from-victoriametrics) when `--vm-concurrency` exceeds 2. Previously the number of idle connections was limited to 2 per host, which was insufficient when `--vm-concurrency` was bigger than 2.
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)
 

docs/victoriametrics/integrations/grafana/_index.md

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

docs/victoriametrics/vmalert.md

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

lib/querytracer/otlp.go

@@ -0,0 +1,122 @@
+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

@@ -0,0 +1,156 @@
+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

@@ -0,0 +1,233 @@
+// 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
+}