app/vmselect: export traces to VictoriaTraces

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/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 {

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
+}