feat: allow span filtering in service graph processor

docs/sources/tempo/configuration/_index.md

@@ -2013,7 +2013,15 @@ overrides:
           [peer_attributes: <list of string>]
           [enable_client_server_prefix: <bool>]
           [enable_messaging_system_latency_histogram: <bool>]
-
+          [filter_policies: [
+            [
+              include/include_any/exclude:
+                match_type: <string> # options: strict, regexp
+                attributes:
+                  - key: <string>
+                    value: <any>
+            ]
+          ]]
         # Configuration for the span-metrics processor
         span_metrics:
           [histogram_buckets: <list of float>]
@@ -2028,7 +2036,7 @@ overrides:
                   - key: <string>
                     value: <any>
             ]
-          ]
+          ]]
           [dimension_mappings: <list of map>]
           # Enable target_info metrics
           [enable_target_info: <bool>]

docs/sources/tempo/configuration/manifest.md

@@ -595,6 +595,7 @@ metrics_generator:
                 - db.namespace
                 - db.name
                 - db.system
+            filter_policies: []
         span_metrics:
             histogram_buckets:
                 - 0.002

docs/sources/tempo/operations/manage-advanced-systems/user-configurable-overrides.md

@@ -84,7 +84,15 @@ metrics_generator:
       [enable_messaging_system_latency_histogram: <bool>]
       [enable_virtual_node_label: <bool>]
       [span_multiplier_key: <string>]
-
+      [filter_policies: [
+        [
+          include/include_any/exclude:
+            match_type: <string> # options: strict, regexp
+            attributes:
+              - key: <string>
+                value: <any>
+        ]
+      ]]
     span_metrics:
       [histogram_buckets: <list of float>]
       [dimensions: <list of string>]

modules/generator/config.go

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

modules/generator/config_test.go

@@ -147,6 +147,65 @@ func TestProcessorConfig_copyWithOverrides(t *testing.T) {
 		}, copied.SpanMetrics.FilterPolicies)
 	})
 
+	t.Run("servicegraphs nil policy overrides", func(t *testing.T) {
+		o := &mockOverrides{
+			serviceGraphsFilterPolicies: nil,
+		}
+
+		copied, err := original.copyWithOverrides(o, "tenant")
+		require.NoError(t, err)
+
+		assert.Equal(t, *original, copied)
+	})
+
+	t.Run("servicegraphs empty policy overrides", func(t *testing.T) {
+		o := &mockOverrides{
+			serviceGraphsFilterPolicies: []config.FilterPolicy{},
+		}
+
+		copied, err := original.copyWithOverrides(o, "tenant")
+		require.NoError(t, err)
+
+		assert.NotEqual(t, *original, copied)
+		assert.Equal(t, []config.FilterPolicy{}, copied.ServiceGraphs.FilterPolicies)
+	})
+
+	t.Run("servicegraphs policy overrides", func(t *testing.T) {
+		o := &mockOverrides{
+			serviceGraphsFilterPolicies: []config.FilterPolicy{
+				{
+					Exclude: &config.PolicyMatch{
+						MatchType: config.Strict,
+						Attributes: []config.MatchPolicyAttribute{
+							{
+								Key:   "resource.service.name",
+								Value: "mythical-requester",
+							},
+						},
+					},
+				},
+			},
+		}
+
+		copied, err := original.copyWithOverrides(o, "tenant")
+		require.NoError(t, err)
+
+		assert.NotEqual(t, *original, copied)
+		assert.Equal(t, []config.FilterPolicy{
+			{
+				Exclude: &config.PolicyMatch{
+					MatchType: config.Strict,
+					Attributes: []config.MatchPolicyAttribute{
+						{
+							Key:   "resource.service.name",
+							Value: "mythical-requester",
+						},
+					},
+				},
+			},
+		}, copied.ServiceGraphs.FilterPolicies)
+	})
+
 	t.Run("span multiplier key overrides", func(t *testing.T) {
 		o := &mockOverrides{
 			serviceGraphsSpanMultiplierKey: "custom_key",

modules/generator/instance.go

@@ -71,6 +71,7 @@
 const (
 	reasonOutsideTimeRangeSlack = "outside_metrics_ingestion_slack"
 	reasonSpanMetricsFiltered   = "span_metrics_filtered"
+	reasonServiceGraphsFiltered = "service_graphs_filtered"
 	reasonInvalidUTF8           = "invalid_utf8"
 )
 
@@ -338,8 +339,12 @@
 			return err
 		}
 	case processor.ServiceGraphsName:
+		filteredSpansCounter := metricSpansDiscarded.WithLabelValues(i.instanceID, reasonServiceGraphsFiltered, processor.ServiceGraphsName)
 		invalidUTF8Counter := metricSpansDiscarded.WithLabelValues(i.instanceID, reasonInvalidUTF8, processor.ServiceGraphsName)
-		newProcessor = servicegraphs.New(cfg.ServiceGraphs, i.instanceID, i.registry, i.logger, invalidUTF8Counter)
+		newProcessor, err = servicegraphs.New(cfg.ServiceGraphs, i.instanceID, i.registry, i.logger, filteredSpansCounter, invalidUTF8Counter)
+		if err != nil {
+			return err
+		}
 	case processor.LocalBlocksName:
 		p, err := localblocks.New(cfg.LocalBlocks, i.instanceID, i.traceWAL, i.writer, i.overrides)
 		if err != nil {

modules/generator/overrides.go

@@ -22,6 +22,7 @@ type metricsGeneratorOverrides interface {
 	MetricsGeneratorProcessorServiceGraphsHistogramBuckets(userID string) []float64
 	MetricsGeneratorProcessorServiceGraphsDimensions(userID string) []string
 	MetricsGeneratorProcessorServiceGraphsPeerAttributes(userID string) []string
+	MetricsGeneratorProcessorServiceGraphsFilterPolicies(userID string) []filterconfig.FilterPolicy
 	MetricsGeneratorProcessorSpanMetricsHistogramBuckets(userID string) []float64
 	MetricsGeneratorProcessorSpanMetricsDimensions(userID string) []string
 	MetricsGeneratorProcessorSpanMetricsIntrinsicDimensions(userID string) map[string]bool

modules/generator/overrides_test.go

@@ -17,6 +17,7 @@ type mockOverrides struct {
 	serviceGraphsHistogramBuckets                      []float64
 	serviceGraphsDimensions                            []string
 	serviceGraphsPeerAttributes                        []string
+	serviceGraphsFilterPolicies                        []filterconfig.FilterPolicy
 	serviceGraphsEnableClientServerPrefix              bool
 	serviceGraphsEnableMessagingSystemLatencyHistogram *bool
 	serviceGraphsEnableVirtualNodeLabel                *bool
@@ -95,6 +96,10 @@ func (m *mockOverrides) MetricsGeneratorProcessorServiceGraphsPeerAttributes(str
 	return m.serviceGraphsPeerAttributes
 }
 
+func (m *mockOverrides) MetricsGeneratorProcessorServiceGraphsFilterPolicies(string) []filterconfig.FilterPolicy {
+	return m.serviceGraphsFilterPolicies
+}
+
 func (m *mockOverrides) MetricsGeneratorProcessorSpanMetricsHistogramBuckets(string) []float64 {
 	return m.spanMetricsHistogramBuckets
 }

modules/generator/processor/servicegraphs/config.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/grafana/tempo/modules/generator/registry"
+	filterconfig "github.com/grafana/tempo/pkg/spanfilter/config"
 	"github.com/prometheus/client_golang/prometheus"
 	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 	semconvnew "go.opentelemetry.io/otel/semconv/v1.34.0"
@@ -50,6 +51,9 @@ type Config struct {
 
 	// DatabaseNameAttributes the list of attribute names used to identify the database name from span attributes.
 	DatabaseNameAttributes []string `yaml:"database_name_attributes"`
+
+	// FilterPolicies is a list of policies that will be applied to spans for inclusion or exclusion.
+	FilterPolicies []filterconfig.FilterPolicy `yaml:"filter_policies"`
 }
 
 func (cfg *Config) RegisterFlagsAndApplyDefaults(string, *flag.FlagSet) {

modules/generator/processor/servicegraphs/servicegraphs.go

@@ -21,6 +21,7 @@
 	"github.com/grafana/tempo/modules/generator/registry"
 	"github.com/grafana/tempo/modules/generator/validation"
 	"github.com/grafana/tempo/pkg/cache/reclaimable"
+	"github.com/grafana/tempo/pkg/spanfilter"
 	"github.com/grafana/tempo/pkg/tempopb"
 	v1_common "github.com/grafana/tempo/pkg/tempopb/common/v1"
 	v1_trace "github.com/grafana/tempo/pkg/tempopb/trace/v1"
@@ -33,6 +34,16 @@
 		Name:      "metrics_generator_processor_service_graphs_dropped_spans",
 		Help:      "Number of spans dropped when trying to add edges",
 	}, []string{"tenant"})
+	metricDroppedEdges = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "tempo",
+		Name:      "metrics_generator_processor_service_graphs_dropped_edges_total",
+		Help:      "Number of edges dropped due to matching a dropped span side counterpart",
+	}, []string{"tenant"})
+	metricDroppedSpanSideCacheOverflows = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "tempo",
+		Name:      "metrics_generator_processor_service_graphs_dropped_span_side_cache_overflow_total",
+		Help:      "Number of dropped span side cache insertions skipped because the cache reached max items",
+	}, []string{"tenant"})
 	metricTotalEdges = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "tempo",
 		Name:      "metrics_generator_processor_service_graphs_edges",
@@ -81,20 +92,28 @@
 	serviceGraphRequestClientSecondsHistogram          registry.Histogram
 	serviceGraphRequestMessagingSystemSecondsHistogram registry.Histogram
 	sanitizeCache                                      reclaimable.Cache[string, string]
-
-	metricDroppedSpans prometheus.Counter
-	metricTotalEdges   prometheus.Counter
-	metricExpiredEdges prometheus.Counter
-	invalidUTF8Counter prometheus.Counter
-	logger             log.Logger
+	filter                                             *spanfilter.SpanFilter
+
+	filteredSpansCounter                prometheus.Counter
+	metricDroppedSpans                  prometheus.Counter
+	metricDroppedEdges                  prometheus.Counter
+	metricDroppedSpanSideCacheOverflows prometheus.Counter
+	metricTotalEdges                    prometheus.Counter
+	metricExpiredEdges                  prometheus.Counter
+	invalidUTF8Counter                  prometheus.Counter
+	logger                              log.Logger
 }
 
-func New(cfg Config, tenant string, reg registry.Registry, logger log.Logger, invalidUTF8Counter prometheus.Counter) gen.Processor {
+func New(cfg Config, tenant string, reg registry.Registry, logger log.Logger, filteredSpansCounter, invalidUTF8Counter prometheus.Counter) (gen.Processor, error) {
 	if cfg.EnableVirtualNodeLabel {
 		cfg.Dimensions = append(cfg.Dimensions, virtualNodeLabel)
 	}
 
 	sanitizeCache := reclaimable.New(validation.SanitizeLabelName, 10000)
+	filter, err := spanfilter.NewSpanFilter(cfg.FilterPolicies)
+	if err != nil {
+		return nil, err
+	}
 
 	p := &Processor{
 		Cfg:      cfg,
@@ -107,15 +126,19 @@
 		serviceGraphRequestClientSecondsHistogram:          reg.NewHistogram(metricRequestClientSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
 		serviceGraphRequestMessagingSystemSecondsHistogram: reg.NewHistogram(metricRequestMessagingSystemSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
 		sanitizeCache: sanitizeCache,
-
-		metricDroppedSpans: metricDroppedSpans.WithLabelValues(tenant),
-		metricTotalEdges:   metricTotalEdges.WithLabelValues(tenant),
-		metricExpiredEdges: metricExpiredEdges.WithLabelValues(tenant),
-		invalidUTF8Counter: invalidUTF8Counter,
-		logger:             log.With(logger, "component", "service-graphs"),
+		filter:        filter,
+
+		filteredSpansCounter:                filteredSpansCounter,
+		metricDroppedSpans:                  metricDroppedSpans.WithLabelValues(tenant),
+		metricDroppedEdges:                  metricDroppedEdges.WithLabelValues(tenant),
+		metricDroppedSpanSideCacheOverflows: metricDroppedSpanSideCacheOverflows.WithLabelValues(tenant),
+		metricTotalEdges:                    metricTotalEdges.WithLabelValues(tenant),
+		metricExpiredEdges:                  metricExpiredEdges.WithLabelValues(tenant),
+		invalidUTF8Counter:                  invalidUTF8Counter,
+		logger:                              log.With(logger, "component", "service-graphs"),
 	}
 
-	p.store = store.NewStore(cfg.Wait, cfg.MaxItems, p.onComplete, p.onExpire)
+	p.store = store.NewStore(cfg.Wait, cfg.MaxItems, p.onComplete, p.onExpire, p.metricDroppedSpanSideCacheOverflows)
 
 	expirationTicker := time.NewTicker(2 * time.Second)
 	for i := 0; i < cfg.Workers; i++ {
@@ -133,7 +156,7 @@
 		}()
 	}
 
-	return p
+	return p, nil
 }
 
 func (p *Processor) Name() string {
@@ -165,6 +188,17 @@
 
 		for _, ils := range rs.ScopeSpans {
 			for _, span := range ils.Spans {
+				// Non-edge spans are ignored by this processor, so skip filter evaluation too.
+				if !isClient(span.Kind) && !isServer(span.Kind) {
+					continue
+				}
+
+				if !p.filter.ApplyFilterPolicy(rs.Resource, span) {
+					p.addDroppedSpanSide(span)
+					p.filteredSpansCounter.Inc()
+					continue
+				}
+
 				connectionType := store.Unknown
 				spanMultiplier := processor_util.GetSpanMultiplier(p.Cfg.SpanMultiplierKey, span, rs.Resource)
 				switch span.Kind {
@@ -174,7 +208,7 @@
 					fallthrough
 				case v1_trace.Span_SPAN_KIND_CLIENT:
 					key := buildKey(hex.EncodeToString(span.TraceId), hex.EncodeToString(span.SpanId))
-					isNew, err = p.store.UpsertEdge(key, func(e *store.Edge) {
+					isNew, err = p.store.UpsertEdge(key, store.Client, func(e *store.Edge) {
 						e.TraceID = tempo_util.TraceIDToHexString(span.TraceId)
 						e.ConnectionType = connectionType
 						e.ClientService = svcName
@@ -193,7 +227,7 @@
 					fallthrough
 				case v1_trace.Span_SPAN_KIND_SERVER:
 					key := buildKey(hex.EncodeToString(span.TraceId), hex.EncodeToString(span.ParentSpanId))
-					isNew, err = p.store.UpsertEdge(key, func(e *store.Edge) {
+					isNew, err = p.store.UpsertEdge(key, store.Server, func(e *store.Edge) {
 						e.TraceID = tempo_util.TraceIDToHexString(span.TraceId)
 						e.ConnectionType = connectionType
 						e.ServerService = svcName
@@ -204,19 +238,17 @@
 						e.SpanMultiplier = spanMultiplier
 						p.upsertPeerNode(e, span.Attributes)
 					})
-				default:
-					// this span is not part of an edge
-					continue
 				}
 
-				if errors.Is(err, store.ErrTooManyItems) {
+				switch {
+				case errors.Is(err, store.ErrTooManyItems):
 					totalDroppedSpans++
 					p.metricDroppedSpans.Inc()
 					continue
-				}
-
-				// UpsertEdge will only return ErrTooManyItems
-				if err != nil {
+				case errors.Is(err, store.ErrDroppedSpanSide):
+					p.metricDroppedEdges.Inc()
+					continue
+				case err != nil:
 					return err
 				}
 
@@ -415,6 +447,36 @@
 	}
 }
 
+func (p *Processor) addDroppedSpanSide(span *v1_trace.Span) {
+	if isClient(span.Kind) {
+		key := buildKey(hex.EncodeToString(span.TraceId), hex.EncodeToString(span.SpanId))
+		if p.store.AddDroppedSpanSide(key, store.Client) {
+			p.metricDroppedEdges.Inc()
+		}
+		return
+	}
+
+	if isServer(span.Kind) {
+		// Root server spans have no parent span ID and cannot match a client counterpart.
+		if len(span.ParentSpanId) == 0 {
+			return
+		}
+
+		key := buildKey(hex.EncodeToString(span.TraceId), hex.EncodeToString(span.ParentSpanId))
+		if p.store.AddDroppedSpanSide(key, store.Server) {
+			p.metricDroppedEdges.Inc()
+		}
+	}
+}
+
+func isClient(kind v1_trace.Span_SpanKind) bool {
+	return kind == v1_trace.Span_SPAN_KIND_CLIENT || kind == v1_trace.Span_SPAN_KIND_PRODUCER
+}
+
+func isServer(kind v1_trace.Span_SpanKind) bool {
+	return kind == v1_trace.Span_SPAN_KIND_SERVER || kind == v1_trace.Span_SPAN_KIND_CONSUMER
+}
+
 func (p *Processor) spanFailed(span *v1_trace.Span) bool {
 	return span.GetStatus().GetCode() == v1_trace.Status_STATUS_CODE_ERROR
 }