Support multiple process handling in otelsdk (#2016)

* Support multiple process handling in otelsdk

* Add example

* Add a changelog entry

Author: MrAlias

CHANGELOG.md

@@ -32,6 +32,8 @@ OpenTelemetry Go Automatic Instrumentation adheres to [Semantic Versioning](http
 - Cache offsets for Go `1.24.2`. ([#2081](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/2081))
 - Cache offsets for `google.golang.org/grpc` `1.73.0-dev`. ([#2091](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/2091))
 - Cache offsets for `golang.org/x/net` `0.39.0`. ([#2107](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/2107))
+- The new `Multiplexer` type is added to `go.opentelemetry.io/auto/pipeline/otelsdk`.
+  This type is used to support multiple process instrumentation using the same telemetry pipeline. ([#2016](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/2016))
 
 ### Changed
 

pipeline/otelsdk/config.go

@@ -172,13 +172,17 @@ type config struct {
 	logger   *slog.Logger
 	exporter sdk.SpanExporter
 	resAttrs []attribute.KeyValue
+
+	spanProcessor sdk.SpanProcessor
+	idGenerator   *idGenerator
 }
 
 func newConfig(ctx context.Context, options []Option) (config, error) {
 	c := config{
 		resAttrs: []attribute.KeyValue{
 			semconv.ServiceName(defaultServiceName()),
 		},
+		idGenerator: newIDGenerator(),
 	}
 
 	var err error
@@ -188,6 +192,15 @@ func newConfig(ctx context.Context, options []Option) (config, error) {
 		err = errors.Join(err, e)
 	}
 
+	if c.exporter == nil {
+		var e error
+		c.exporter, e = otlptracehttp.New(ctx)
+		if e != nil {
+			err = errors.Join(err, e)
+		}
+	}
+	c.spanProcessor = sdk.NewBatchSpanProcessor(c.exporter)
+
 	return c, err
 }
 
@@ -206,24 +219,15 @@ func (c config) Logger() *slog.Logger {
 	return newLogger(nil)
 }
 
-func (c config) TracerProvider(ctx context.Context) (*sdk.TracerProvider, error) {
-	exp := c.exporter
-	if exp == nil {
-		var err error
-		exp, err = otlptracehttp.New(ctx)
-		if err != nil {
-			return nil, err
-		}
-	}
-
+func (c config) TracerProvider() *sdk.TracerProvider {
 	return sdk.NewTracerProvider(
 		// Sample everything. The actual sampling is done in the eBPF probes
 		// before it reaches this tracerProvider.
 		sdk.WithSampler(sdk.AlwaysSample()),
 		sdk.WithResource(c.resource()),
-		sdk.WithBatcher(exp),
-		sdk.WithIDGenerator(newIDGenerator()),
-	), nil
+		sdk.WithSpanProcessor(c.spanProcessor),
+		sdk.WithIDGenerator(c.idGenerator),
+	)
 }
 
 func (c config) resource() *resource.Resource {

pipeline/otelsdk/example_test.go

@@ -0,0 +1,61 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package otelsdk_test
+
+import (
+	"context"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"go.opentelemetry.io/auto"
+	"go.opentelemetry.io/auto/pipeline/otelsdk"
+)
+
+func Example_multiplex() {
+	// Create a context that cancels when a SIGTERM is received. This ensures
+	// that each instrumentation goroutine below can shut down cleanly.
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
+	defer stop()
+
+	// Create a new multiplexer to handle instrumentation events from multiple
+	// sources. This will act as a central router for telemetry handlers.
+	m, err := otelsdk.NewMultiplexer(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	// Simulated process IDs to be instrumented. These would typically be real
+	// process IDs in a production scenario.
+	pids := []int{1297, 1331, 9827}
+
+	var wg sync.WaitGroup
+	for _, pid := range pids {
+		wg.Add(1)
+
+		go func(id int) {
+			defer wg.Done()
+
+			// Create a new instrumentation session for the process.
+			//
+			// NOTE: Error handling is omitted here for brevity. In production
+			// code, always check and handle errors.
+			inst, _ := auto.NewInstrumentation(
+				ctx,
+				auto.WithPID(id),
+				auto.WithHandler(m.Handler(id)),
+			)
+
+			// Load and start the instrumentation for the process.
+			_ = inst.Load(ctx)
+			_ = inst.Run(ctx)
+		}(pid)
+	}
+
+	// Wait for all instrumentation goroutines to complete.
+	wg.Wait()
+
+	// Shut down the multiplexer, cleaning up any remaining resources.
+	_ = m.Shutdown(ctx)
+}

pipeline/otelsdk/handler.go

@@ -53,12 +53,11 @@ func NewTraceHandler(ctx context.Context, options ...Option) (*TraceHandler, err
 		return nil, err
 	}
 
-	tp, err := c.TracerProvider(ctx)
-	if err != nil {
-		return nil, err
-	}
+	return newTraceHandler(c), nil
+}
 
-	return &TraceHandler{logger: c.Logger(), tracerProvider: tp}, nil
+func newTraceHandler(c config) *TraceHandler {
+	return &TraceHandler{logger: c.Logger(), tracerProvider: c.TracerProvider()}
 }
 
 // HandleTrace the passed telemetry using the default OpenTelemetry Go SDK.

pipeline/otelsdk/multiplex.go

@@ -0,0 +1,103 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package otelsdk
+
+import (
+	"context"
+	"debug/buildinfo"
+	"strconv"
+
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.30.0"
+
+	"go.opentelemetry.io/auto/pipeline"
+)
+
+// Multiplexer supports sending telemetry from multiple resources through the
+// same processing and exporting pipeline.
+type Multiplexer struct {
+	cfg config
+}
+
+// NewMultiplexer returns a new *Multiplexer that reuses the provided options as
+// a base configuration for all [pipeline.Handler] it creates.
+//
+// Any exporter provided in the options must support exporting telemetry for
+// multiple resources.
+func NewMultiplexer(ctx context.Context, options ...Option) (*Multiplexer, error) {
+	cfg, err := newConfig(ctx, options)
+	if err != nil {
+		return nil, err
+	}
+	return &Multiplexer{cfg: cfg}, nil
+}
+
+// Handler returns a new [pipeline.Handler] configured with additional
+// resource attributes that follow OpenTelemetry semantic conventions for the
+// process associated with the given pid.
+//
+// If an error occurs while determining process-specific resource attributes,
+// the error is logged, and a handler without those attributes is returned.
+//
+// If Shutdown has already been called on the Multiplexer, the returned handler
+// will also be in a shut down state and will not export any telemetry.
+func (m Multiplexer) Handler(pid int) *pipeline.Handler {
+	c := m.withProcResAttrs(pid)
+	return &pipeline.Handler{TraceHandler: newTraceHandler(c)}
+}
+
+// Shutdown gracefully shuts down the Multiplexer's span processor.
+//
+// After Shutdown is called, any subsequent calls to Handler will return a
+// handler that is in a shut down state. These handlers will silently drop
+// telemetry and will not perform any processing or exporting.
+func (m Multiplexer) Shutdown(ctx context.Context) error {
+	return m.cfg.spanProcessor.Shutdown(ctx)
+}
+
+// withProcResAttrs returns a copy of the Multiplexer's config with additional
+// resource attributes based on Go runtime build information for the process
+// identified by pid.
+//
+// It attempts to read the build info of the Go executable at /proc/<pid>/exe
+// and extracts semantic convention attributes such as the runtime version and
+// compiler name. If this fails, an error is logged and no extra attributes
+// are added.
+func (m Multiplexer) withProcResAttrs(pid int) (c config) {
+	c = m.cfg // Make a shallow copy to modify attributes.
+
+	var attrs []attribute.KeyValue
+
+	path := "/proc/" + strconv.Itoa(pid) + "/exe"
+	bi, err := buildinfo.ReadFile(path)
+	if err != nil {
+		c.logger.Error("failed to get Go proc build info", "error", err)
+		return c
+	}
+
+	// Add Go runtime version as a semantic attribute.
+	attrs = append(attrs, semconv.ProcessRuntimeVersion(bi.GoVersion))
+
+	// Try to determine which Go compiler was used.
+	var compiler string
+	for _, setting := range bi.Settings {
+		if setting.Key == "-compiler" {
+			compiler = setting.Value
+			break
+		}
+	}
+	switch compiler {
+	case "":
+		c.logger.Debug("failed to identify Go compiler")
+	case "gc":
+		attrs = append(attrs, semconv.ProcessRuntimeName("go"))
+	default:
+		attrs = append(attrs, semconv.ProcessRuntimeName(compiler))
+	}
+
+	// Prepend process-specific attributes so user-provided ones have priority.
+	c.resAttrs = append(attrs, c.resAttrs...)
+
+	return c
+}