feature: implemented jaeger service dependencies API

app/victoria-traces/main.go

@@ -15,6 +15,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
 
+	"github.com/VictoriaMetrics/VictoriaTraces/app/victoria-traces/servicegraph"
 	"github.com/VictoriaMetrics/VictoriaTraces/app/vtinsert"
 	"github.com/VictoriaMetrics/VictoriaTraces/app/vtinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaTraces/app/vtselect"
@@ -49,6 +50,9 @@ func main() {
 	insertutil.SetLogRowsStorage(&vtstorage.Storage{})
 	vtinsert.Init()
 
+	// optional background task(s)
+	servicegraph.Init()
+
 	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
 		UseProxyProtocol: useProxyProtocol,
 	})
@@ -66,6 +70,7 @@ func main() {
 	}
 	logger.Infof("successfully shut down the webservice in %.3f seconds", time.Since(startTime).Seconds())
 
+	servicegraph.Stop()
 	vtinsert.Stop()
 	vtselect.Stop()
 	vtstorage.Stop()

app/victoria-traces/servicegraph/servicegraph.go

@@ -0,0 +1,101 @@
+package servicegraph
+
+import (
+	"context"
+	"flag"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+
+	vtinsert "github.com/VictoriaMetrics/VictoriaTraces/app/vtinsert/opentelemetry"
+	vtselect "github.com/VictoriaMetrics/VictoriaTraces/app/vtselect/traces/query"
+	"github.com/VictoriaMetrics/VictoriaTraces/app/vtstorage"
+)
+
+var (
+	enableServiceGraphTask     = flag.Bool("servicegraph.enableTask", false, "Whether to enable background task for generating service graph. It should only be enabled on VictoriaTraces single-node or vtstorage.")
+	serviceGraphTaskInterval   = flag.Duration("servicegraph.taskInterval", time.Minute, "The background task interval for generating service graph data. It requires setting -servicegraph.enableTask=true.")
+	serviceGraphTaskTimeout    = flag.Duration("servicegraph.taskTimeout", 30*time.Second, "The background task timeout duration for generating service graph data. It requires setting -servicegraph.enableTask=true.")
+	serviceGraphTaskLookbehind = flag.Duration("servicegraph.taskLookbehind", time.Minute, "The lookbehind window for each time service graph background task run. It requires setting -servicegraph.enableTask=true.")
+	serviceGraphTaskLimit      = flag.Uint64("servicegraph.taskLimit", 1000, "How many service graph relations each task could fetch for each tenant. It requires setting -servicegraph.enableTask=true.")
+)
+
+var (
+	sgt *serviceGraphTask
+)
+
+func Init() {
+	if *enableServiceGraphTask {
+		sgt = newServiceGraphTask()
+		sgt.Start()
+	}
+}
+
+func Stop() {
+	if *enableServiceGraphTask {
+		sgt.Stop()
+	}
+}
+
+type serviceGraphTask struct {
+	stopCh chan struct{}
+}
+
+func newServiceGraphTask() *serviceGraphTask {
+	return &serviceGraphTask{
+		stopCh: make(chan struct{}),
+	}
+}
+
+func (sgt *serviceGraphTask) Start() {
+	logger.Infof("starting servicegraph background task, interval: %v, lookbehind: %v", *serviceGraphTaskInterval, *serviceGraphTaskLookbehind)
+	go func() {
+		ticker := time.NewTicker(*serviceGraphTaskInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-sgt.stopCh:
+				return
+			case <-ticker.C:
+				ctx, cancelFunc := context.WithTimeout(context.Background(), *serviceGraphTaskTimeout)
+				GenerateServiceGraphTimeRange(ctx)
+				cancelFunc()
+			}
+		}
+	}()
+}
+
+func (sgt *serviceGraphTask) Stop() {
+	close(sgt.stopCh)
+}
+
+func GenerateServiceGraphTimeRange(ctx context.Context) {
+	endTime := time.Now().Truncate(*serviceGraphTaskInterval)
+	startTime := endTime.Add(-*serviceGraphTaskLookbehind)
+
+	tenantIDs, err := vtstorage.GetTenantIDsByTimeRange(ctx, startTime.UnixNano(), endTime.UnixNano())
+	if err != nil {
+		logger.Errorf("cannot get tenant ids: %s", err)
+		return
+	}
+
+	// query and persist operations are executed sequentially, which helps not to consume excessive resources.
+	for _, tenantID := range tenantIDs {
+		// query service graph relations
+		rows, err := vtselect.GetServiceGraphTimeRange(ctx, tenantID, startTime, endTime, *serviceGraphTaskLimit)
+		if err != nil {
+			logger.Errorf("cannot get service graph for time range [%d, %d]: %s", startTime.Unix(), endTime.Unix(), err)
+			return
+		}
+		if len(rows) == 0 {
+			return
+		}
+
+		// persist service graph relations
+		err = vtinsert.PersistServiceGraph(ctx, tenantID, rows, endTime)
+		if err != nil {
+			logger.Errorf("cannot presist service graph for time range [%d, %d]: %s", startTime.Unix(), endTime.Unix(), err)
+		}
+	}
+}

app/vtgen/main.go

@@ -68,6 +68,7 @@ func main() {
 				// The traceIDMap recorded old traceID->new traceID.
 				// Spans with same old traceID should be replaced with same new traceID.
 				traceIDMap := make(map[string]string)
+				spanIDMap := make(map[string]string)
 
 				// The timeOffset is the time offset of span timestamp and current timestamp.
 				// All spans' timestamp should be increased by this offset.
@@ -115,6 +116,26 @@ func main() {
 									}
 								}
 
+								// replace SpanID
+								if sid, ok := spanIDMap[sp.SpanID]; ok {
+									sp.SpanID = sid
+								} else {
+									spanID := generateSpanID()
+									oldSpanID := sp.SpanID
+									sp.SpanID = spanID
+									spanIDMap[oldSpanID] = spanID
+								}
+
+								// replace parentSpanID
+								if sid, ok := spanIDMap[sp.ParentSpanID]; ok {
+									sp.ParentSpanID = sid
+								} else {
+									parentSpanID := generateSpanID()
+									oldParentSpanID := sp.ParentSpanID
+									sp.ParentSpanID = parentSpanID
+									spanIDMap[oldParentSpanID] = parentSpanID
+								}
+
 								// adjust the timestamp of the span.
 								sp.StartTimeUnixNano = sp.StartTimeUnixNano + timeOffset
 								sp.EndTimeUnixNano = sp.EndTimeUnixNano + timeOffset + uint64(rand.Int63n(100000000))
@@ -198,12 +219,27 @@ func loadTestData() [][]byte {
 	return bodyList
 }
 
+var traceIDMutex sync.Mutex
+
 func generateTraceID() string {
+	traceIDMutex.Lock()
+	defer traceIDMutex.Unlock()
+
 	h := md5.New()
 	h.Write([]byte(strconv.FormatInt(time.Now().UnixNano(), 10)))
 	return hex.EncodeToString(h.Sum(nil))
 }
 
+var spanIDMutex sync.Mutex
+
+func generateSpanID() string {
+	spanIDMutex.Lock()
+	defer spanIDMutex.Unlock()
+	h := md5.New()
+	h.Write([]byte(strconv.FormatInt(time.Now().UnixNano(), 10)))
+	return hex.EncodeToString(h.Sum(nil))[:16]
+}
+
 // readWrite Does the following:
 // 1. read request body binary files like `1.bin`, `2.bin` and puts them into `BodyList`.
 // 2. encode and compress the `BodyList` into `[]byte`.

app/vtinsert/opentelemetry/opentelemetry.go

@@ -1,6 +1,7 @@
 package opentelemetry
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -284,3 +285,21 @@ func appendKeyValuesWithPrefixSuffix(fields []logstorage.Field, kvs []*otelpb.Ke
 	}
 	return fields
 }
+
+func PersistServiceGraph(ctx context.Context, tenantID logstorage.TenantID, fields [][]logstorage.Field, timestamp time.Time) error {
+	cp := insertutil.CommonParams{
+		TenantID:   tenantID,
+		TimeFields: []string{"_time"},
+	}
+	lmp := cp.NewLogMessageProcessor("internalinsert_servicegraph", false)
+
+	for _, row := range fields {
+		f := append(row, logstorage.Field{
+			Name:  "_msg",
+			Value: "-",
+		})
+		lmp.AddRow(timestamp.UnixNano(), f, []logstorage.Field{{Name: otelpb.ServiceGraphStreamName, Value: "-"}})
+	}
+	lmp.MustClose()
+	return nil
+}

app/vtselect/traces/jaeger/jaeger.go

@@ -73,8 +73,7 @@ func RequestHandler(ctx context.Context, w http.ResponseWriter, r *http.Request)
 		return true
 	} else if path == "/select/jaeger/api/dependencies" {
 		jaegerDependenciesRequests.Inc()
-		// todo it require additional component to calculate the dependency graph. not implemented yet.
-		httpserver.Errorf(w, r, "/api/dependencies API is not supported yet.")
+		processGetDependenciesRequest(ctx, w, r)
 		jaegerDependenciesDuration.UpdateDuration(startTime)
 		return true
 	}
@@ -401,3 +400,91 @@ func hashProcess(process process) uint64 {
 	hashpool.Put(d)
 	return h
 }
+
+// processGetDependenciesRequest handle the Jaeger /api/dependencies API request.
+func processGetDependenciesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	cp, err := query.GetCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "incorrect query params: %s", err)
+		return
+	}
+
+	param, err := parseJaegerDependenciesQueryParam(ctx, r)
+	if err != nil {
+		httpserver.Errorf(w, r, "incorrect dependencies query params: %s", err)
+		return
+	}
+
+	rows, err := query.GetServiceGraphList(ctx, cp, param)
+	if err != nil {
+		httpserver.Errorf(w, r, "get dependencies error: %s", err)
+		return
+	}
+
+	if len(rows) == 0 {
+		// Write empty results
+		w.Header().Set("Content-Type", "application/json")
+		WriteGetDependenciesResponse(w, nil)
+		return
+	}
+
+	dependencies := make([]*dependencyLink, 0)
+	for _, row := range rows {
+		dependency := &dependencyLink{}
+		for _, f := range row.Fields {
+			switch f.Name {
+			case "parent":
+				dependency.parent = f.Value
+			case "child":
+				dependency.child = f.Value
+			case "callCount":
+				dependency.callCount, err = strconv.ParseUint(f.Value, 10, 64)
+				if err != nil {
+					logger.Errorf("cannot parse callCount [%s]: %s", f.Value, err)
+					continue
+				}
+			}
+		}
+		if dependency.parent != "" && dependency.child != "" && dependency.callCount > 0 {
+			dependencies = append(dependencies, dependency)
+		}
+	}
+
+	// Write results
+	w.Header().Set("Content-Type", "application/json")
+	WriteGetDependenciesResponse(w, dependencies)
+}
+
+// parseJaegerDependenciesQueryParam parse Jaeger request to unified ServiceGraphQueryParameters.
+func parseJaegerDependenciesQueryParam(_ context.Context, r *http.Request) (*query.ServiceGraphQueryParameters, error) {
+	var err error
+
+	// default params
+	p := &query.ServiceGraphQueryParameters{
+		EndTs:    time.Now(),
+		Lookback: time.Hour,
+	}
+	q := r.URL.Query()
+
+	endTs := q.Get("endTs")
+	if endTs != "" {
+		unixMilli, err := strconv.ParseInt(endTs, 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse endTs [%s]: %w", endTs, err)
+		}
+		p.EndTs = time.UnixMilli(unixMilli)
+	}
+
+	lookback := q.Get("lookback")
+	if lookback != "" {
+		if strings.TrimLeft(lookback, "0123456789") == "" {
+			lookback += "ms"
+		}
+		p.Lookback, err = time.ParseDuration(lookback)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse lookback [%s]: %w", lookback, err)
+		}
+	}
+
+	return p, nil
+}

app/vtselect/traces/jaeger/jaeger.qtpl

@@ -63,6 +63,31 @@
 }
 {% endfunc %}
 
+{% func GetDependenciesResponse(dependencies []*dependencyLink) %}
+{
+	"data":[
+        {% if len(dependencies) > 0 %}
+            {%= dependencyJson(dependencies[0]) %}
+            {% for _, dependency := range dependencies[1:] %}
+                ,{%= dependencyJson(dependency) %}
+            {% endfor %}
+        {% endif %}
+	],
+	"errors": null,
+	"limit": 0,
+	"offset": 0,
+	"total": {%d= len(dependencies) %}
+}
+{% endfunc %}
+
+{% func dependencyJson(dependency *dependencyLink) %}
+{
+    "parent": {%q= dependency.parent %},
+    "child": {%q= dependency.child %},
+    "callCount": {%dul= dependency.callCount %}
+}
+{% endfunc %}
+
 {% func traceJson(trace *trace) %}
 {
     "processes": {