Pass telemetry directly to the Controller

Do not enqueue telemetry to a channel just to dequeue it and send it 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

@@ -235,7 +235,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),
 	}
 
@@ -288,7 +287,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 {
@@ -308,7 +307,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 = true
 }
 
@@ -370,7 +369,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},
@@ -475,10 +473,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 {
@@ -498,7 +496,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))
 	}
 }