Add ability to detect virtual nodes in the servicegraph processor

CHANGELOG.md

@@ -30,6 +30,7 @@ To make use of filtering, configure `autocomplete_filtering_enabled`.
 * [ENHANCEMENT] Add option to override metrics-generator ring port  [#2399](https://github.com/grafana/tempo/pull/2399) (@mdisibio)
 * [ENHANCEMENT] Add support for IPv6 [#1555](https://github.com/grafana/tempo/pull/1555) (@zalegrala)
 * [ENHANCEMENT] Add span filtering to spanmetrics processor [#2274](https://github.com/grafana/tempo/pull/2274) (@zalegrala)
+* [ENHANCEMENT] Add ability to detect virtual nodes in the servicegraph processor [#2365](https://github.com/grafana/tempo/pull/2365) (@mapno)
 * [BUGFIX] tempodb integer divide by zero error [#2167](https://github.com/grafana/tempo/issues/2167) (@kroksys)
 * [BUGFIX] metrics-generator: ensure Prometheus will scale up shards when remote write is lagging behind [#2463](https://github.com/grafana/tempo/issues/2463) (@kvrhdn)
 * [CHANGE] **Breaking Change** Rename s3.insecure_skip_verify [#2407](https://github.com/grafana/tempo/pull/2407) (@zalegrala)

cmd/tempo-serverless/cloud-run/go.mod

@@ -119,7 +119,7 @@ require (
 	github.com/willf/bloom v2.0.3+incompatible // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.0.0-rc8 // indirect
-	go.opentelemetry.io/collector/semconv v0.74.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.75.0 // indirect
 	go.opentelemetry.io/otel v1.14.0 // indirect
 	go.opentelemetry.io/otel/trace v1.14.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect

cmd/tempo-serverless/cloud-run/go.sum

@@ -547,8 +547,8 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/collector/pdata v1.0.0-rc8 h1:vBikWdZFsRiT5dVsLQhnE99w3edM7eem3Q9dSqMlStE=
 go.opentelemetry.io/collector/pdata v1.0.0-rc8/go.mod h1:BVCBhWgclYCh7Oi6BkMiQfRa6MXv1uRTlKXuL5oBby8=
-go.opentelemetry.io/collector/semconv v0.74.0 h1:tPpbz87CPu/pM2/fSEKBJWXTvWvUJvEChbQkzdhWQHE=
-go.opentelemetry.io/collector/semconv v0.74.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
+go.opentelemetry.io/collector/semconv v0.75.0 h1:zIlZk+zh1bgc3VKE1PZEmhOaVa4tQHZMcFFUXmGekVs=
+go.opentelemetry.io/collector/semconv v0.75.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
 go.opentelemetry.io/otel v1.14.0 h1:/79Huy8wbf5DnIPhemGB+zEPVwnN6fuQybr/SRXa6hM=
 go.opentelemetry.io/otel v1.14.0/go.mod h1:o4buv+dJzx8rohcUeRmWUZhqupFvzWis188WlggnNeU=
 go.opentelemetry.io/otel/trace v1.14.0 h1:wp2Mmvj41tDsyAJXiWDWpfNsOiIyd38fy85pyKcFq/M=

cmd/tempo-serverless/lambda/go.mod

@@ -123,7 +123,7 @@ require (
 	github.com/willf/bloom v2.0.3+incompatible // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.0.0-rc8 // indirect
-	go.opentelemetry.io/collector/semconv v0.74.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.75.0 // indirect
 	go.opentelemetry.io/otel v1.14.0 // indirect
 	go.opentelemetry.io/otel/trace v1.14.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect

cmd/tempo-serverless/lambda/go.sum

@@ -556,8 +556,8 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/collector/pdata v1.0.0-rc8 h1:vBikWdZFsRiT5dVsLQhnE99w3edM7eem3Q9dSqMlStE=
 go.opentelemetry.io/collector/pdata v1.0.0-rc8/go.mod h1:BVCBhWgclYCh7Oi6BkMiQfRa6MXv1uRTlKXuL5oBby8=
-go.opentelemetry.io/collector/semconv v0.74.0 h1:tPpbz87CPu/pM2/fSEKBJWXTvWvUJvEChbQkzdhWQHE=
-go.opentelemetry.io/collector/semconv v0.74.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
+go.opentelemetry.io/collector/semconv v0.75.0 h1:zIlZk+zh1bgc3VKE1PZEmhOaVa4tQHZMcFFUXmGekVs=
+go.opentelemetry.io/collector/semconv v0.75.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
 go.opentelemetry.io/otel v1.14.0 h1:/79Huy8wbf5DnIPhemGB+zEPVwnN6fuQybr/SRXa6hM=
 go.opentelemetry.io/otel v1.14.0/go.mod h1:o4buv+dJzx8rohcUeRmWUZhqupFvzWis188WlggnNeU=
 go.opentelemetry.io/otel/trace v1.14.0 h1:wp2Mmvj41tDsyAJXiWDWpfNsOiIyd38fy85pyKcFq/M=

cmd/tempo/main.go

@@ -30,7 +30,7 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/sdk/resource"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
-	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
 	"gopkg.in/yaml.v2"
 )
 

docs/sources/tempo/configuration/_index.md

@@ -1252,6 +1252,7 @@ overrides:
     # overrides settings in the global configuration.
     [metrics_generator_processor_service_graphs_histogram_buckets: <list of float>]
     [metrics_generator_processor_service_graphs_dimensions: <list of string>]
+    [metrics_generator_processor_service_graphs_peer_attributes: <list of string>]
     [metrics_generator_processor_span_metrics_histogram_buckets: <list of float>]
     # Allowed keys for intrinsic dimensions are: service, span_name, span_kind, status_code, and status_message.
     [metrics_generator_processor_span_metrics_intrinsic_dimensions: <map string to bool>]

docs/sources/tempo/metrics-generator/service_graphs.md

@@ -43,6 +43,18 @@ Each emitted metrics series have the `client` and `server` label corresponding w
   tempo_service_graph_request_total{client="app", server="db", connection_type="database"} 20
 ```
 
+#### Virtual nodes
+
+Virtual nodes are nodes that form part of the lifecycle of a trace,
+but spans for them are not being collected because they're outside the user's reach (for example, an external service for payment processing) or are not instrumented (for example, a frontend application).
+
+Virtual nodes can be detected in two different ways:
+
+- The root span has `span.kind` set to `server`. This indicates that the request has initiated by an external system that's not instrumented, like a frontend application or an engineer via `curl`.
+- A `client` span does not have its matching `server` span, but has a peer attribute present. In this case, we make the assumption that a call was made to an external service, for which Tempo won't receive spans.
+   - The default peer attributes are `peer.service`, `net.peer.name`, `net.sock.peer.name`, `rpc.service.key`, `net.sock.peer.addr`, `http.url`, `http.target`.
+   - The order of the attributes is important, as the first one that is present will be used as the virtual node name.
+
 ### Metrics
 
 The following metrics are exported:

go.mod

@@ -77,7 +77,7 @@ require (
 	go.opentelemetry.io/collector/exporter/otlpexporter v0.74.0
 	go.opentelemetry.io/collector/pdata v1.0.0-rc8
 	go.opentelemetry.io/collector/receiver/otlpreceiver v0.74.0
-	go.opentelemetry.io/collector/semconv v0.74.0
+	go.opentelemetry.io/collector/semconv v0.75.0
 	go.opentelemetry.io/otel v1.14.0
 	go.opentelemetry.io/otel/bridge/opencensus v0.37.0
 	go.opentelemetry.io/otel/bridge/opentracing v1.10.0

go.sum

@@ -1042,8 +1042,8 @@ go.opentelemetry.io/collector/receiver v0.74.0 h1:jlgBFa0iByvn8VuX27UxtqiPiZE8ej
 go.opentelemetry.io/collector/receiver v0.74.0/go.mod h1:SQkyATvoZCJefNkI2jnrR63SOdrmDLYCnQqXJ7ACqn0=
 go.opentelemetry.io/collector/receiver/otlpreceiver v0.74.0 h1:e/X/W0z2Jtpy3Yd3CXkmEm9vSpKq/P3pKUrEVMUFBRw=
 go.opentelemetry.io/collector/receiver/otlpreceiver v0.74.0/go.mod h1:9X9/RYFxJIaK0JLlRZ0PpmQSSlYpY+r4KsTOj2jWj14=
-go.opentelemetry.io/collector/semconv v0.74.0 h1:tPpbz87CPu/pM2/fSEKBJWXTvWvUJvEChbQkzdhWQHE=
-go.opentelemetry.io/collector/semconv v0.74.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
+go.opentelemetry.io/collector/semconv v0.75.0 h1:zIlZk+zh1bgc3VKE1PZEmhOaVa4tQHZMcFFUXmGekVs=
+go.opentelemetry.io/collector/semconv v0.75.0/go.mod h1:xt8oDOiwa1jy24tGUo8+SzpphI7ZredS2WM/0m8rtTA=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.40.0 h1:5jD3teb4Qh7mx/nfzq4jO2WFFpvXD0vYWFDrdvNWmXk=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.40.0/go.mod h1:UMklln0+MRhZC4e3PwmN3pCtq4DyIadWw4yikh6bNrw=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.40.0 h1:lE9EJyw3/JhrjWH/hEy9FptnalDQgj7vpbgC2KCCCxE=

modules/generator/config.go

@@ -69,6 +69,9 @@ func (cfg *ProcessorConfig) copyWithOverrides(o metricsGeneratorOverrides, userI
 	if dimensions := o.MetricsGeneratorProcessorServiceGraphsDimensions(userID); dimensions != nil {
 		copyCfg.ServiceGraphs.Dimensions = dimensions
 	}
+	if peerAttrs := o.MetricsGeneratorProcessorServiceGraphsPeerAttributes(userID); peerAttrs != nil {
+		copyCfg.ServiceGraphs.PeerAttributes = peerAttrs
+	}
 	if buckets := o.MetricsGeneratorProcessorSpanMetricsHistogramBuckets(userID); buckets != nil {
 		copyCfg.SpanMetrics.HistogramBuckets = buckets
 	}

modules/generator/overrides.go

@@ -15,6 +15,7 @@ type metricsGeneratorOverrides interface {
 	MetricsGeneratorProcessors(userID string) map[string]struct{}
 	MetricsGeneratorProcessorServiceGraphsHistogramBuckets(userID string) []float64
 	MetricsGeneratorProcessorServiceGraphsDimensions(userID string) []string
+	MetricsGeneratorProcessorServiceGraphsPeerAttributes(userID string) []string
 	MetricsGeneratorProcessorSpanMetricsHistogramBuckets(userID string) []float64
 	MetricsGeneratorProcessorSpanMetricsDimensions(userID string) []string
 	MetricsGeneratorProcessorSpanMetricsIntrinsicDimensions(userID string) map[string]bool

modules/generator/overrides_test.go

@@ -11,6 +11,7 @@ type mockOverrides struct {
 	processors                      map[string]struct{}
 	serviceGraphsHistogramBuckets   []float64
 	serviceGraphsDimensions         []string
+	serviceGraphsPeerAttributes     []string
 	spanMetricsHistogramBuckets     []float64
 	spanMetricsDimensions           []string
 	spanMetricsIntrinsicDimensions  map[string]bool
@@ -51,6 +52,10 @@ func (m *mockOverrides) MetricsGeneratorProcessorServiceGraphsDimensions(userID
 	return m.serviceGraphsDimensions
 }
 
+func (m *mockOverrides) MetricsGeneratorProcessorServiceGraphsPeerAttributes(userID string) []string {
+	return m.serviceGraphsPeerAttributes
+}
+
 func (m *mockOverrides) MetricsGeneratorProcessorSpanMetricsHistogramBuckets(userID string) []float64 {
 	return m.spanMetricsHistogramBuckets
 }

modules/generator/processor/servicegraphs/config.go

@@ -28,6 +28,10 @@ type Config struct {
 	// (either value could get used)
 	Dimensions []string `yaml:"dimensions"`
 
+	// PeerAttributes are attributes that will be used to create a peer edge
+	// Attributes are searched in the order they are provided
+	PeerAttributes []string `yaml:"peer_attributes"`
+
 	// If enabled attribute value will be used for metric calculation
 	SpanMultiplierKey string `yaml:"span_multiplier_key"`
 }
@@ -38,4 +42,10 @@ func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet)
 	cfg.Workers = 10
 	// TODO: Revisit this default value.
 	cfg.HistogramBuckets = prometheus.ExponentialBuckets(0.1, 2, 8)
+
+	peerAttr := make([]string, 0, len(defaultPeerAttributes))
+	for _, attr := range defaultPeerAttributes {
+		peerAttr = append(peerAttr, string(attr))
+	}
+	cfg.PeerAttributes = peerAttr
 }

modules/generator/processor/servicegraphs/servicegraphs.go

@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/go-kit/log"
@@ -13,6 +14,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/prometheus/util/strutil"
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
 
 	gen "github.com/grafana/tempo/modules/generator/processor"
 	"github.com/grafana/tempo/modules/generator/processor/servicegraphs/store"
@@ -49,6 +52,10 @@ const (
 	metricRequestClientSeconds = "traces_service_graph_request_client_seconds"
 )
 
+var defaultPeerAttributes = []attribute.Key{
+	semconv.PeerServiceKey, semconv.NetPeerNameKey, semconv.NetSockPeerNameKey, semconv.RPCServiceKey, semconv.NetSockPeerAddrKey, semconv.HTTPURLKey, semconv.HTTPTargetKey,
+}
+
 type tooManySpansError struct {
 	droppedSpans int
 }
@@ -169,6 +176,7 @@ func (p *Processor) consume(resourceSpans []*v1_trace.ResourceSpans) (err error)
 						e.Failed = e.Failed || p.spanFailed(span)
 						p.upsertDimensions(e.Dimensions, rs.Resource.Attributes, span.Attributes)
 						e.SpanMultiplier = spanMultiplier
+						p.upsertPeerNode(e, span.Attributes)
 
 						// A database request will only have one span, we don't wait for the server
 						// span but just copy details from the client span
@@ -193,6 +201,7 @@ func (p *Processor) consume(resourceSpans []*v1_trace.ResourceSpans) (err error)
 						e.Failed = e.Failed || p.spanFailed(span)
 						p.upsertDimensions(e.Dimensions, rs.Resource.Attributes, span.Attributes)
 						e.SpanMultiplier = spanMultiplier
+						p.upsertPeerNode(e, span.Attributes)
 					})
 				default:
 					// this span is not part of an edge
@@ -234,6 +243,15 @@ func (p *Processor) upsertDimensions(m map[string]string, resourceAttr []*v1_com
 	}
 }
 
+func (p *Processor) upsertPeerNode(e *store.Edge, spanAttr []*v1_common.KeyValue) {
+	for _, peerKey := range p.Cfg.PeerAttributes {
+		if v, ok := processor_util.FindAttributeValue(peerKey, spanAttr); ok {
+			e.PeerNode = v
+			return
+		}
+	}
+}
+
 func (p *Processor) Shutdown(_ context.Context) {
 	close(p.closeCh)
 }
@@ -259,6 +277,25 @@ func (p *Processor) onComplete(e *store.Edge) {
 
 func (p *Processor) onExpire(e *store.Edge) {
 	p.metricExpiredEdges.Inc()
+
+	// If an edge is expired, we check if there are signs that the missing span is belongs to a "virtual node".
+	// These are nodes that are outside the user's reach (eg. an external service for payment processing),
+	// or that are not instrumented (eg. a frontend application).
+	e.ConnectionType = store.VirtualNode
+	if len(e.ClientService) == 0 {
+		// If the client service is not set, it means that the span could have been initiated by an external system,
+		// like a frontend application or an engineer via `curl`.
+		// We check if the span we have is the root span, and if so, we set the client service to "user".
+		if _, parentSpan := parseKey(e.Key()); len(parentSpan) == 0 {
+			e.ClientService = "user"
+			p.onComplete(e)
+		}
+	} else if len(e.ServerService) == 0 && len(e.PeerNode) > 0 {
+		// If client span does not have its matching server span, but has a peer attribute present,
+		// we make the assumption that a call was made to an external service, for which Tempo won't receive spans.
+		e.ServerService = e.PeerNode
+		p.onComplete(e)
+	}
 }
 
 func (p *Processor) spanFailed(span *v1_trace.Span) bool {
@@ -272,3 +309,8 @@ func spanDurationSec(span *v1_trace.Span) float64 {
 func buildKey(k1, k2 string) string {
 	return fmt.Sprintf("%s-%s", k1, k2)
 }
+
+func parseKey(key string) (string, string) {
+	parts := strings.Split(key, "-")
+	return parts[0], parts[1]
+}

modules/generator/processor/servicegraphs/servicegraphs_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/jsonpb"
@@ -150,6 +151,47 @@ func TestServiceGraphs_tooManySpansErr(t *testing.T) {
 	assert.True(t, errors.As(err, &tooManySpansError{}))
 }
 
+func TestServiceGraphs_virtualNodes(t *testing.T) {
+	testRegistry := registry.NewTestRegistry()
+
+	cfg := Config{}
+	cfg.RegisterFlagsAndApplyDefaults("", nil)
+
+	cfg.HistogramBuckets = []float64{0.04}
+	cfg.Wait = time.Nanosecond
+
+	p := New(cfg, "test", testRegistry, log.NewNopLogger())
+	defer p.Shutdown(context.Background())
+
+	request, err := loadTestData("testdata/trace-with-virtual-nodes.json")
+	require.NoError(t, err)
+
+	p.PushSpans(context.Background(), request)
+
+	p.(*Processor).store.Expire()
+
+	userToServerLabels := labels.FromMap(map[string]string{
+		"client":          "user",
+		"server":          "mythical-server",
+		"connection_type": "virtual_node",
+	})
+
+	clientToVirtualPeerLabels := labels.FromMap(map[string]string{
+		"client":          "mythical-requester",
+		"server":          "external-payments-platform",
+		"connection_type": "virtual_node",
+	})
+
+	fmt.Println(testRegistry)
+
+	// counters
+	assert.Equal(t, 1.0, testRegistry.Query(`traces_service_graph_request_total`, userToServerLabels))
+	assert.Equal(t, 0.0, testRegistry.Query(`traces_service_graph_request_failed_total`, userToServerLabels))
+
+	assert.Equal(t, 1.0, testRegistry.Query(`traces_service_graph_request_total`, clientToVirtualPeerLabels))
+	assert.Equal(t, 0.0, testRegistry.Query(`traces_service_graph_request_failed_total`, clientToVirtualPeerLabels))
+}
+
 func loadTestData(path string) (*tempopb.PushSpansRequest, error) {
 	f, err := os.Open(path)
 	if err != nil {

modules/generator/processor/servicegraphs/store/edge.go

@@ -8,6 +8,7 @@ const (
 	Unknown         ConnectionType = ""
 	MessagingSystem ConnectionType = "messaging_system"
 	Database        ConnectionType = "database"
+	VirtualNode     ConnectionType = "virtual_node"
 )
 
 // Edge is an Edge between two nodes in the graph
@@ -26,6 +27,9 @@ type Edge struct {
 	// Additional dimension to add to the metrics
 	Dimensions map[string]string
 
+	// PeerNode is the attribute that will be used to create a peer edge
+	PeerNode string
+
 	// expiration is the time at which the Edge expires, expressed as Unix time
 	expiration int64
 
@@ -50,3 +54,7 @@ func (e *Edge) isComplete() bool {
 func (e *Edge) isExpired() bool {
 	return time.Now().Unix() >= e.expiration
 }
+
+func (e *Edge) Key() string {
+	return e.key
+}