Pass telemetry directly to the Controller

internal/pkg/instrumentation/manager.go

@@ -13,7 +13,6 @@ import (
 	"github.com/cilium/ebpf/link"
 	"github.com/cilium/ebpf/rlimit"
 
-	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/otel/trace"
 
 	dbSql "go.opentelemetry.io/auto/internal/pkg/instrumentation/bpf/database/sql"
@@ -58,7 +57,6 @@ type Manager struct {
 	exe             *link.Executable
 	td              *process.TargetDetails
 	runningProbesWG sync.WaitGroup
-	telemetryCh     chan ptrace.ScopeSpans
 	currentConfig   Config
 	probeMu         sync.Mutex
 	state           managerState
@@ -74,7 +72,6 @@ func NewManager(logger *slog.Logger, otelController *opentelemetry.Controller, g
 		globalImpl:      globalImpl,
 		loadedIndicator: loadIndicator,
 		cp:              cp,
-		telemetryCh:     make(chan ptrace.ScopeSpans),
 	}
 
 	err := m.registerProbes()
@@ -227,7 +224,7 @@ func (m *Manager) runProbe(p probe.Probe) {
 	m.runningProbesWG.Add(1)
 	go func(ap probe.Probe) {
 		defer m.runningProbesWG.Done()
-		ap.Run(m.telemetryCh)
+		ap.Run(m.otelController.Trace)
 	}(p)
 }
 
@@ -290,19 +287,15 @@ func (m *Manager) Run(ctx context.Context, target *process.TargetDetails) error
 		m.logger.Debug("Shutting down all probes")
 		err := m.cleanup(target)
 
-		// Wait for all probes to stop before closing the chan they send on.
+		// Wait for all probes to stop.
 		m.runningProbesWG.Wait()
-		close(m.telemetryCh)
 
 		m.state = managerStateStopped
 		m.probeMu.Unlock()
 
 		done <- errors.Join(err, ctx.Err())
 	}()
 
-	for e := range m.telemetryCh {
-		m.otelController.Trace(e)
-	}
 	return <-done
 }
 

internal/pkg/instrumentation/manager_test.go

@@ -236,7 +236,6 @@ func TestRunStopping(t *testing.T) {
 		otelController: ctrl,
 		logger:         slog.Default(),
 		probes:         map[probe.ID]probe.Probe{{}: p},
-		telemetryCh:    make(chan ptrace.ScopeSpans),
 		cp:             NewNoopConfigProvider(nil),
 	}
 
@@ -289,7 +288,7 @@ func (p slowProbe) Load(*link.Executable, *process.TargetDetails, *sampling.Conf
 	return nil
 }
 
-func (p slowProbe) Run(c chan<- ptrace.ScopeSpans) {
+func (p slowProbe) Run(func(ptrace.ScopeSpans)) {
 }
 
 func (p slowProbe) Close() error {
@@ -309,7 +308,7 @@ func (p *noopProbe) Load(*link.Executable, *process.TargetDetails, *sampling.Con
 	return nil
 }
 
-func (p *noopProbe) Run(c chan<- ptrace.ScopeSpans) {
+func (p *noopProbe) Run(func(ptrace.ScopeSpans)) {
 	p.running.Store(true)
 }
 
@@ -371,7 +370,6 @@ func TestConfigProvider(t *testing.T) {
 			netHTTPServerProbeID:       &noopProbe{},
 			somePackageProducerProbeID: &noopProbe{},
 		},
-		telemetryCh: make(chan ptrace.ScopeSpans),
 		cp: newDummyProvider(Config{
 			InstrumentationLibraryConfigs: map[LibraryID]Library{
 				netHTTPClientLibID: {TracesEnabled: &falseVal},
@@ -488,10 +486,10 @@ func (p *hangingProbe) Load(*link.Executable, *process.TargetDetails, *sampling.
 	return nil
 }
 
-func (p *hangingProbe) Run(c chan<- ptrace.ScopeSpans) {
+func (p *hangingProbe) Run(handle func(ptrace.ScopeSpans)) {
 	<-p.closeReturned
 	// Write after Close has returned.
-	c <- ptrace.NewScopeSpans()
+	handle(ptrace.NewScopeSpans())
 }
 
 func (p *hangingProbe) Close() error {
@@ -511,7 +509,6 @@ func TestRunStopDeadlock(t *testing.T) {
 		otelController: ctrl,
 		logger:         slog.Default(),
 		probes:         map[probe.ID]probe.Probe{{}: p},
-		telemetryCh:    make(chan ptrace.ScopeSpans),
 		cp:             NewNoopConfigProvider(nil),
 	}
 

internal/pkg/instrumentation/probe/probe.go

@@ -42,7 +42,7 @@ type Probe interface {
 	Load(*link.Executable, *process.TargetDetails, *sampling.Config) error
 
 	// Run runs the events processing loop.
-	Run(tracesChan chan<- ptrace.ScopeSpans)
+	Run(func(ptrace.ScopeSpans))
 
 	// Close stops the Probe.
 	Close() error
@@ -267,7 +267,7 @@ type SpanProducer[BPFObj any, BPFEvent any] struct {
 }
 
 // Run runs the events processing loop.
-func (i *SpanProducer[BPFObj, BPFEvent]) Run(dest chan<- ptrace.ScopeSpans) {
+func (i *SpanProducer[BPFObj, BPFEvent]) Run(handle func(ptrace.ScopeSpans)) {
 	for {
 		event, err := i.read()
 		if err != nil {
@@ -285,7 +285,7 @@ func (i *SpanProducer[BPFObj, BPFEvent]) Run(dest chan<- ptrace.ScopeSpans) {
 
 		i.ProcessFn(event).CopyTo(ss.Spans())
 
-		dest <- ss
+		handle(ss)
 	}
 }
 
@@ -296,7 +296,7 @@ type TraceProducer[BPFObj any, BPFEvent any] struct {
 }
 
 // Run runs the events processing loop.
-func (i *TraceProducer[BPFObj, BPFEvent]) Run(dest chan<- ptrace.ScopeSpans) {
+func (i *TraceProducer[BPFObj, BPFEvent]) Run(handle func(ptrace.ScopeSpans)) {
 	for {
 		event, err := i.read()
 		if err != nil {
@@ -306,7 +306,7 @@ func (i *TraceProducer[BPFObj, BPFEvent]) Run(dest chan<- ptrace.ScopeSpans) {
 			continue
 		}
 
-		dest <- i.ProcessFn(event)
+		handle(i.ProcessFn(event))
 	}
 }
 

internal/pkg/opentelemetry/controller.go

@@ -19,22 +19,11 @@ import (
 type Controller struct {
 	logger         *slog.Logger
 	tracerProvider trace.TracerProvider
-	tracersMap     map[tracerID]trace.Tracer
-}
-
-type tracerID struct{ name, version, schema string }
-
-func (c *Controller) getTracer(name, version, schema string) trace.Tracer {
-	tID := tracerID{name: name, version: version, schema: schema}
-	t, exists := c.tracersMap[tID]
-	if !exists {
-		t = c.tracerProvider.Tracer(name, trace.WithInstrumentationVersion(version), trace.WithSchemaURL(schema))
-		c.tracersMap[tID] = t
-	}
-	return t
 }
 
 // Trace creates a trace span for event.
+//
+// This method is safe to call concurrently.
 func (c *Controller) Trace(ss ptrace.ScopeSpans) {
 	var (
 		startOpts []trace.SpanStartOption
@@ -43,15 +32,20 @@ func (c *Controller) Trace(ss ptrace.ScopeSpans) {
 		kvs       []attribute.KeyValue
 	)
 
-	t := c.getTracer(ss.Scope().Name(), ss.Scope().Version(), ss.SchemaUrl())
+	tracer := c.tracerProvider.Tracer(
+		ss.Scope().Name(),
+		trace.WithInstrumentationVersion(ss.Scope().Version()),
+		trace.WithInstrumentationAttributes(attrs(ss.Scope().Attributes())...),
+		trace.WithSchemaURL(ss.SchemaUrl()),
+	)
 	for k := 0; k < ss.Spans().Len(); k++ {
 		pSpan := ss.Spans().At(k)
 
 		if pSpan.TraceID().IsEmpty() || pSpan.SpanID().IsEmpty() {
 			c.logger.Debug("dropping invalid span", "name", pSpan.Name())
 			continue
 		}
-		c.logger.Debug("handling span", "tracer", t, "span", pSpan)
+		c.logger.Debug("handling span", "tracer", tracer, "span", pSpan)
 
 		ctx := context.Background()
 		if !pSpan.ParentSpanID().IsEmpty() {
@@ -71,7 +65,7 @@ func (c *Controller) Trace(ss ptrace.ScopeSpans) {
 			trace.WithTimestamp(pSpan.StartTimestamp().AsTime()),
 			trace.WithLinks(c.links(pSpan.Links())...),
 		)
-		_, span := t.Start(ctx, pSpan.Name(), startOpts...)
+		_, span := tracer.Start(ctx, pSpan.Name(), startOpts...)
 		startOpts = startOpts[:0]
 		kvs = kvs[:0]
 
@@ -96,7 +90,6 @@ func NewController(logger *slog.Logger, tracerProvider trace.TracerProvider) (*C
 	return &Controller{
 		logger:         logger,
 		tracerProvider: tracerProvider,
-		tracersMap:     make(map[tracerID]trace.Tracer),
 	}, nil
 }
 
@@ -113,6 +106,11 @@ func (c *Controller) Shutdown(ctx context.Context) error {
 	return nil
 }
 
+func attrs(m pcommon.Map) []attribute.KeyValue {
+	out := make([]attribute.KeyValue, 0, m.Len())
+	return appendAttrs(out, m)
+}
+
 func appendAttrs(dest []attribute.KeyValue, m pcommon.Map) []attribute.KeyValue {
 	m.Range(func(k string, v pcommon.Value) bool {
 		dest = append(dest, attr(k, v))
@@ -213,8 +211,7 @@ func appendEventOpts(dest []trace.EventOption, e ptrace.SpanEvent) []trace.Event
 		dest = append(dest, trace.WithTimestamp(ts))
 	}
 
-	var kvs []attribute.KeyValue
-	kvs = appendAttrs(kvs, e.Attributes())
+	kvs := attrs(e.Attributes())
 	if len(kvs) > 0 {
 		dest = append(dest, trace.WithAttributes(kvs...))
 	}
@@ -244,8 +241,8 @@ func (c *Controller) links(links ptrace.SpanLinkSlice) []trace.Link {
 				TraceFlags: trace.TraceFlags(l.Flags()),
 				TraceState: ts,
 			}),
+			Attributes: attrs(l.Attributes()),
 		}
-		out[i].Attributes = appendAttrs(out[i].Attributes, l.Attributes())
 	}
 	return out
 }

internal/pkg/opentelemetry/controller_test.go

@@ -10,6 +10,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -314,35 +315,6 @@ func TestTrace(t *testing.T) {
 	}
 }
 
-func TestGetTracer(t *testing.T) {
-	exporter := tracetest.NewInMemoryExporter()
-	tp := sdktrace.NewTracerProvider(
-		sdktrace.WithSampler(sdktrace.AlwaysSample()),
-		sdktrace.WithBatcher(exporter),
-		sdktrace.WithResource(instResource()),
-	)
-	defer func() {
-		err := tp.Shutdown(context.Background())
-		assert.NoError(t, err)
-	}()
-
-	ctrl, err := NewController(slog.Default(), tp)
-	assert.NoError(t, err)
-
-	t1 := ctrl.getTracer("test", "v1", "schema")
-	assert.Equal(t, t1, ctrl.tracersMap[tracerID{name: "test", version: "v1", schema: "schema"}])
-
-	t2 := ctrl.getTracer("net/http", "", "")
-	assert.Equal(t, t2, ctrl.tracersMap[tracerID{name: "net/http", version: "", schema: ""}])
-
-	t3 := ctrl.getTracer("test", "v1", "schema")
-	assert.Same(t, t1, t3)
-
-	t4 := ctrl.getTracer("net/http", "", "")
-	assert.Same(t, t2, t4)
-	assert.Equal(t, len(ctrl.tracersMap), 2)
-}
-
 type shutdownExporter struct {
 	sdktrace.SpanExporter
 
@@ -390,3 +362,38 @@ func TestShutdown(t *testing.T) {
 	assert.True(t, exporter.called, "Exporter not shutdown")
 	assert.Equal(t, uint32(nSpan), exporter.exported.Load(), "Pending spans not flushed")
 }
+
+func TestControllerTraceConcurrentSafe(t *testing.T) {
+	exporter := tracetest.NewInMemoryExporter()
+	tp := sdktrace.NewTracerProvider(
+		sdktrace.WithSampler(sdktrace.AlwaysSample()),
+		sdktrace.WithBatcher(exporter),
+		sdktrace.WithResource(instResource()),
+	)
+	defer func() {
+		err := tp.Shutdown(context.Background())
+		assert.NoError(t, err)
+	}()
+
+	ctrl, err := NewController(slog.Default(), tp)
+	assert.NoError(t, err)
+
+	const goroutines = 10
+
+	var wg sync.WaitGroup
+	for n := 0; n < goroutines; n++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			data := ptrace.NewScopeSpans()
+			data.Scope().SetName(fmt.Sprintf("tracer-%d", n%(goroutines/2)))
+			data.Scope().SetVersion("v1")
+			data.SetSchemaUrl("url")
+			data.Spans().AppendEmpty().SetName("test")
+			ctrl.Trace(data)
+		}()
+	}
+
+	wg.Wait()
+}