feat: adds plugin span filtering for multiple connectors

core/schemas/span_filter.go

@@ -0,0 +1,118 @@
+package schemas
+
+import (
+	"fmt"
+	"slices"
+	"strings"
+)
+
+// PluginSpanFilterMode controls whether the plugins list is an allowlist or denylist.
+type PluginSpanFilterMode string
+
+const (
+	// PluginSpanFilterModeInclude exports only the listed plugins' spans.
+	PluginSpanFilterModeInclude PluginSpanFilterMode = "include"
+	// PluginSpanFilterModeExclude exports everything except the listed plugins' spans.
+	PluginSpanFilterModeExclude PluginSpanFilterMode = "exclude"
+)
+
+// PluginSpanFilter configures which plugin spans an observability connector exports.
+// Mode "include" exports only the listed plugins; mode "exclude" exports everything
+// except them. It is shared by every observability connector (OTEL, Datadog, BigQuery)
+// so the span-name contract and reparenting behavior stay consistent across exporters.
+type PluginSpanFilter struct {
+	Mode    PluginSpanFilterMode `json:"mode"`
+	Plugins []string             `json:"plugins"`
+}
+
+// Validate reports whether the filter's mode is one of the two valid modes.
+// A nil filter is valid (it filters nothing).
+func (f *PluginSpanFilter) Validate() error {
+	if f == nil {
+		return nil
+	}
+	switch f.Mode {
+	case PluginSpanFilterModeInclude, PluginSpanFilterModeExclude:
+		return nil
+	default:
+		return fmt.Errorf("plugin_span_filter.mode %q is invalid: must be %q or %q",
+			f.Mode, PluginSpanFilterModeInclude, PluginSpanFilterModeExclude)
+	}
+}
+
+// PluginNameFromSpan extracts "<name>" from a plugin span whose name follows the
+// core tracer contract "plugin.<name>.<stage>", where <stage> is one of prehook,
+// posthook, mcp_prehook, mcp_posthook, mcp_connect_prehook, or mcp_connect_posthook
+// (see core/bifrost.go). It returns "" for non-plugin spans or names that don't match
+// the contract (wrong prefix, or fewer than three segments), so malformed names pass
+// through ShouldExportSpan as exported rather than being silently filtered.
+//
+// The <stage> segment is intentionally not constrained to a fixed list: the tracer
+// emits several hook stages (including the mcp_* variants above), so pinning it to
+// just prehook/posthook would make every MCP-hook span unfilterable.
+func PluginNameFromSpan(span *Span) string {
+	if span == nil || span.Kind != SpanKindPlugin {
+		return ""
+	}
+	parts := strings.SplitN(span.Name, ".", 3)
+	if len(parts) != 3 || parts[0] != "plugin" || parts[1] == "" {
+		return ""
+	}
+	return parts[1]
+}
+
+// ShouldExportSpan reports whether a span survives the filter. Non-plugin spans and
+// spans evaluated against a nil filter are always exported. Plugin spans are checked
+// against the filter's plugin list and mode.
+func (f *PluginSpanFilter) ShouldExportSpan(span *Span) bool {
+	if f == nil || span == nil || span.Kind != SpanKindPlugin {
+		return true
+	}
+	pluginName := PluginNameFromSpan(span)
+	if pluginName == "" {
+		// Malformed plugin span name: export rather than silently drop.
+		return true
+	}
+	inList := slices.Contains(f.Plugins, pluginName)
+	if f.Mode == PluginSpanFilterModeInclude {
+		return inList
+	}
+	return !inList // exclude mode
+}
+
+// BuildReparentMap returns a map of filteredSpanID → effective ancestor spanID for all
+// spans that the filter removes. When plugin spans are chained (each span's parent is the
+// previous plugin's span), removing a span from the middle would leave its children with a
+// dangling parent ID. The map lets callers rewrite those parent IDs to the nearest exported
+// ancestor, handling consecutive filtered spans in a chain. Returns nil when the filter is
+// nil or nothing is filtered.
+func (f *PluginSpanFilter) BuildReparentMap(spans []*Span) map[string]string {
+	if f == nil {
+		return nil
+	}
+	// First pass: record direct parent ID for every filtered span.
+	filtered := make(map[string]string) // spanID -> parentID
+	for _, span := range spans {
+		if !f.ShouldExportSpan(span) {
+			filtered[span.SpanID] = span.ParentID
+		}
+	}
+	if len(filtered) == 0 {
+		return nil
+	}
+	// Second pass: resolve chains so each filtered span maps to its first exported ancestor.
+	// Cap the walk at len(filtered) to break out of any cycle caused by malformed span data.
+	maxHops := len(filtered)
+	for spanID := range filtered {
+		parentID := filtered[spanID]
+		for range maxHops {
+			grandParentID, isFiltered := filtered[parentID]
+			if !isFiltered {
+				break
+			}
+			parentID = grandParentID
+		}
+		filtered[spanID] = parentID
+	}
+	return filtered
+}

core/schemas/span_filter_test.go

@@ -0,0 +1,143 @@
+package schemas
+
+import "testing"
+
+func pluginSpan(id, parent, name string) *Span {
+	return &Span{SpanID: id, ParentID: parent, Name: name, Kind: SpanKindPlugin}
+}
+
+func TestPluginSpanFilter_Validate(t *testing.T) {
+	tests := []struct {
+		name    string
+		filter  *PluginSpanFilter
+		wantErr bool
+	}{
+		{"nil filter", nil, false},
+		{"include", &PluginSpanFilter{Mode: PluginSpanFilterModeInclude}, false},
+		{"exclude", &PluginSpanFilter{Mode: PluginSpanFilterModeExclude}, false},
+		{"invalid mode", &PluginSpanFilter{Mode: "nonsense"}, true},
+		{"empty mode", &PluginSpanFilter{Mode: ""}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := tt.filter.Validate(); (err != nil) != tt.wantErr {
+				t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestPluginNameFromSpan(t *testing.T) {
+	tests := []struct {
+		name string
+		span *Span
+		want string
+	}{
+		{"prehook", pluginSpan("1", "", "plugin.logging.prehook"), "logging"},
+		{"posthook", pluginSpan("1", "", "plugin.compat.posthook"), "compat"},
+		{"mcp hook stage still resolves", pluginSpan("1", "", "plugin.governance.mcp_connect_prehook"), "governance"},
+		{"non-plugin kind", &Span{Name: "plugin.logging.prehook", Kind: SpanKindLLMCall}, ""},
+		{"malformed name", pluginSpan("1", "", "plugin"), ""},
+		{"missing stage", pluginSpan("1", "", "plugin.logging"), ""},
+		{"wrong prefix", pluginSpan("1", "", "otel.logging.prehook"), ""},
+		{"empty name segment", pluginSpan("1", "", "plugin..prehook"), ""},
+		{"nil span", nil, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := PluginNameFromSpan(tt.span); got != tt.want {
+				t.Errorf("PluginNameFromSpan() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestPluginSpanFilter_ShouldExportSpan(t *testing.T) {
+	llm := &Span{SpanID: "llm", Name: "llm.call", Kind: SpanKindLLMCall}
+	logging := pluginSpan("p1", "", "plugin.logging.prehook")
+	compat := pluginSpan("p2", "", "plugin.compat.prehook")
+
+	tests := []struct {
+		name   string
+		filter *PluginSpanFilter
+		span   *Span
+		want   bool
+	}{
+		{"nil filter exports plugin", nil, logging, true},
+		{"non-plugin always exported", &PluginSpanFilter{Mode: PluginSpanFilterModeInclude, Plugins: []string{"logging"}}, llm, true},
+		{"include lists plugin", &PluginSpanFilter{Mode: PluginSpanFilterModeInclude, Plugins: []string{"logging"}}, logging, true},
+		{"include omits plugin", &PluginSpanFilter{Mode: PluginSpanFilterModeInclude, Plugins: []string{"logging"}}, compat, false},
+		{"exclude lists plugin", &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"logging"}}, logging, false},
+		{"exclude omits plugin", &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"logging"}}, compat, true},
+		{"malformed plugin span exported", &PluginSpanFilter{Mode: PluginSpanFilterModeInclude, Plugins: []string{"logging"}}, pluginSpan("p3", "", "plugin"), true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.filter.ShouldExportSpan(tt.span); got != tt.want {
+				t.Errorf("ShouldExportSpan() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestPluginSpanFilter_BuildReparentMap(t *testing.T) {
+	t.Run("nil filter returns nil", func(t *testing.T) {
+		f := (*PluginSpanFilter)(nil)
+		if got := f.BuildReparentMap([]*Span{pluginSpan("1", "", "plugin.logging.prehook")}); got != nil {
+			t.Errorf("expected nil, got %v", got)
+		}
+	})
+
+	t.Run("nothing filtered returns nil", func(t *testing.T) {
+		f := &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"absent"}}
+		spans := []*Span{pluginSpan("1", "", "plugin.logging.prehook")}
+		if got := f.BuildReparentMap(spans); got != nil {
+			t.Errorf("expected nil, got %v", got)
+		}
+	})
+
+	t.Run("single filtered span maps to its parent", func(t *testing.T) {
+		// root(llm) <- logging <- compat. Exclude logging only.
+		f := &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"logging"}}
+		spans := []*Span{
+			{SpanID: "root", Name: "llm.call", Kind: SpanKindLLMCall},
+			pluginSpan("logging", "root", "plugin.logging.prehook"),
+			pluginSpan("compat", "logging", "plugin.compat.prehook"),
+		}
+		got := f.BuildReparentMap(spans)
+		if got["logging"] != "root" {
+			t.Errorf("logging should reparent to root, got %q", got["logging"])
+		}
+	})
+
+	t.Run("chain of filtered spans resolves to first exported ancestor", func(t *testing.T) {
+		// root(llm) <- a <- b <- c. Exclude a and b. c should reparent to root.
+		f := &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"a", "b"}}
+		spans := []*Span{
+			{SpanID: "root", Name: "llm.call", Kind: SpanKindLLMCall},
+			pluginSpan("a", "root", "plugin.a.prehook"),
+			pluginSpan("b", "a", "plugin.b.prehook"),
+			pluginSpan("c", "b", "plugin.c.prehook"),
+		}
+		got := f.BuildReparentMap(spans)
+		if got["a"] != "root" {
+			t.Errorf("a should resolve to root, got %q", got["a"])
+		}
+		if got["b"] != "root" {
+			t.Errorf("b should resolve to root, got %q", got["b"])
+		}
+		if _, ok := got["c"]; ok {
+			t.Errorf("c is exported and should not be in the map")
+		}
+	})
+
+	t.Run("cycle is bounded and does not hang", func(t *testing.T) {
+		// Malformed: a's parent is b, b's parent is a. Both filtered.
+		f := &PluginSpanFilter{Mode: PluginSpanFilterModeExclude, Plugins: []string{"a", "b"}}
+		spans := []*Span{
+			pluginSpan("a", "b", "plugin.a.prehook"),
+			pluginSpan("b", "a", "plugin.b.prehook"),
+		}
+		_ = f.BuildReparentMap(spans) // must terminate
+	})
+}

plugins/otel/converter.go

@@ -3,7 +3,6 @@ package otel
 import (
 	"encoding/hex"
 	"fmt"
-	"slices"
 	"strings"
 
 	"github.com/maximhq/bifrost/core/schemas"
@@ -70,72 +69,15 @@ func hexToBytes(hexStr string, length int) []byte {
 	return bytes
 }
 
-// shouldExportSpan reports whether a span should be included in the export.
-// Non-plugin spans are always exported. Plugin spans are checked against pluginSpanFilter.
-func (p *OtelPlugin) shouldExportSpan(span *schemas.Span) bool {
-	if span.Kind != schemas.SpanKindPlugin || p.pluginSpanFilter == nil {
-		return true
-	}
-	// Span names follow the pattern "plugin.<name>.prehook" / "plugin.<name>.posthook".
-	parts := strings.SplitN(span.Name, ".", 3)
-	if len(parts) < 2 {
-		return true
-	}
-	pluginName := parts[1]
-
-	inList := slices.Contains(p.pluginSpanFilter.Plugins, pluginName)
-
-	if p.pluginSpanFilter.Mode == PluginSpanFilterModeInclude {
-		return inList
-	}
-	return !inList // exclude mode
-}
-
-// buildReparentMap returns a map of filteredSpanID → effective ancestor spanID for all
-// spans that will be skipped. When plugin spans are chained (each span's parent is the
-// previous plugin's span), removing a span from the middle would leave its children with
-// a dangling parent ID. The map lets us rewrite those parent IDs to the nearest exported
-// ancestor, handling consecutive filtered spans in a chain.
-func (p *OtelPlugin) buildReparentMap(spans []*schemas.Span) map[string]string {
-	if p.pluginSpanFilter == nil {
-		return nil
-	}
-	// First pass: record direct parent ID for every filtered span.
-	filtered := make(map[string]string) // spanID -> parentID
-	for _, span := range spans {
-		if !p.shouldExportSpan(span) {
-			filtered[span.SpanID] = span.ParentID
-		}
-	}
-	if len(filtered) == 0 {
-		return nil
-	}
-	// Second pass: resolve chains so each filtered span maps to its first exported ancestor.
-	// Cap the walk at len(filtered) to break out of any cycle caused by malformed span data.
-	maxHops := len(filtered)
-	for spanID := range filtered {
-		parentID := filtered[spanID]
-		for range maxHops {
-			grandParentID, isFiltered := filtered[parentID]
-			if !isFiltered {
-				break
-			}
-			parentID = grandParentID
-		}
-		filtered[spanID] = parentID
-	}
-	return filtered
-}
-
 // convertTraceToResourceSpan converts a Bifrost trace to OTEL ResourceSpan for the given
 // profile service name. Span filtering and instance attributes are shared across profiles;
 // only the resource service name differs per profile.
 func (p *OtelPlugin) convertTraceToResourceSpan(serviceName string, trace *schemas.Trace, requestHeaders []string, disableContentLogging bool) *ResourceSpan {
-	reparent := p.buildReparentMap(trace.Spans)
+	reparent := p.pluginSpanFilter.BuildReparentMap(trace.Spans)
 	filteredHeaders := schemas.FilterHeaders(trace.RequestHeaders, requestHeaders)
 	otelSpans := make([]*Span, 0, len(trace.Spans))
 	for _, span := range trace.Spans {
-		if !p.shouldExportSpan(span) {
+		if !p.pluginSpanFilter.ShouldExportSpan(span) {
 			continue
 		}
 		otelSpan := convertSpanToOTELSpan(trace.TraceID, span, disableContentLogging)