Introduce Config Provider as an option to set the initial configurati…

…on and update it. (#1010)

* Add config.Provider to load/unload probes

* Added comments, improve log message and unit test

* update test file

* changleog entry

* rename config fields to be traces specific

* Hadle potential races between applying config and closing

* clarigy shutdown function

---------

Co-authored-by: Eden Federman <eden@keyval.dev>

CHANGELOG.md

@@ -15,6 +15,7 @@ OpenTelemetry Go Automatic Instrumentation adheres to [Semantic Versioning](http
 - Support `golang.org/x/net` `v0.28.0`. ([#988](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/988))
 - Support `google.golang.org/grpc` `1.67.0-dev`. ([#1007](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1007))
 - Support Go `1.23.0`.  ([#1007](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1007))
+- Introduce `config.Provider` as an option to set the initial configuration and update it in runtime. ([#1010](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/1010))
 
 ### Fixed
 

config/provider.go

@@ -0,0 +1,74 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/trace"
+)
+
+// InstrumentationLibraryID is used to identify an instrumentation library.
+type InstrumentationLibraryID struct {
+	// Name of the instrumentation pkg (e.g. "net/http").
+	InstrumentedPkg string
+	// SpanKind is the relevant span kind for the instrumentation.
+	// This can be used to configure server-only, client-only spans.
+	// If not set, the identifier is assumed to be applicable to all span kinds relevant to the instrumented package.
+	SpanKind trace.SpanKind
+}
+
+// InstrumentationLibrary is used to configure instrumentation for a specific library.
+type InstrumentationLibrary struct {
+	// TracesEnabled determines whether traces are enabled for the instrumentation library.
+	// if nil - take DefaultTracesDisabled value.
+	TracesEnabled *bool
+}
+
+// InstrumentationConfig is used to configure instrumentation.
+type InstrumentationConfig struct {
+	// InstrumentationLibraryConfigs defines library-specific configuration.
+	// If a package is referenced by more than one key, the most specific key is used.
+	// For example, if ("net/http", unspecified) and ("net/http", client) are both present,
+	// the configuration for ("net/http", client) is used for client spans and the configuration for ("net/http", unspecified) is used for server spans.
+	InstrumentationLibraryConfigs map[InstrumentationLibraryID]InstrumentationLibrary
+
+	// DefaultTracesDisabled determines whether traces are disabled by default.
+	// If set to true, traces are disabled by default for all libraries, unless the library is explicitly enabled.
+	// If set to false, traces are enabled by default for all libraries, unless the library is explicitly disabled.
+	// default is false - traces are enabled by default.
+	DefaultTracesDisabled bool
+}
+
+// Provider provides the initial configuration and updates to the instrumentation configuration.
+type Provider interface {
+	// InitialConfig returns the initial instrumentation configuration.
+	InitialConfig(ctx context.Context) InstrumentationConfig
+	// Watch returns a channel that receives updates to the instrumentation configuration.
+	Watch() <-chan InstrumentationConfig
+	// Shutdown releases any resources held by the provider.
+	// It is an error to send updates after Shutdown is called.
+	Shutdown(ctx context.Context) error
+}
+
+type noopProvider struct{}
+
+// NewNoopProvider returns a provider that does not provide any updates and provide the default configuration as the initial one.
+func NewNoopProvider() Provider {
+	return &noopProvider{}
+}
+
+func (p *noopProvider) InitialConfig(_ context.Context) InstrumentationConfig {
+	return InstrumentationConfig{}
+}
+
+func (p *noopProvider) Watch() <-chan InstrumentationConfig {
+	c := make(chan InstrumentationConfig)
+	close(c)
+	return c
+}
+
+func (p *noopProvider) Shutdown(_ context.Context) error {
+	return nil
+}

instrumentation.go

@@ -28,6 +28,7 @@ import (
 	"go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 
+	"go.opentelemetry.io/auto/config"
 	"go.opentelemetry.io/auto/internal/pkg/instrumentation"
 	"go.opentelemetry.io/auto/internal/pkg/opentelemetry"
 	"go.opentelemetry.io/auto/internal/pkg/process"
@@ -74,11 +75,11 @@ func newLogger(logLevel LogLevel) logr.Logger {
 		level, _ = zap.ParseAtomicLevel(LogLevelInfo.String())
 	}
 
-	config := zap.NewProductionConfig()
+	c := zap.NewProductionConfig()
 
-	config.Level.SetLevel(level.Level())
+	c.Level.SetLevel(level.Level())
 
-	zapLog, err := config.Build()
+	zapLog, err := c.Build()
 
 	var logger logr.Logger
 	if err != nil {
@@ -130,7 +131,7 @@ func NewInstrumentation(ctx context.Context, opts ...InstrumentationOption) (*In
 		return nil, err
 	}
 
-	mngr, err := instrumentation.NewManager(logger, ctrl, c.globalImpl, c.loadIndicator)
+	mngr, err := instrumentation.NewManager(logger, ctrl, c.globalImpl, c.loadIndicator, c.cp)
 	if err != nil {
 		return nil, err
 	}
@@ -221,6 +222,7 @@ type instConfig struct {
 	globalImpl         bool
 	loadIndicator      chan struct{}
 	logLevel           LogLevel
+	cp                 config.Provider
 }
 
 func newInstConfig(ctx context.Context, opts []InstrumentationOption) (instConfig, error) {
@@ -255,6 +257,10 @@ func newInstConfig(ctx context.Context, opts []InstrumentationOption) (instConfi
 		c.logLevel = LogLevelInfo
 	}
 
+	if c.cp == nil {
+		c.cp = config.NewNoopProvider()
+	}
+
 	return c, err
 }
 
@@ -562,3 +568,14 @@ func WithLogLevel(level LogLevel) InstrumentationOption {
 		return c, nil
 	})
 }
+
+// WithConfigProvider returns an [InstrumentationOption] that will configure
+// an [Instrumentation] to use the provided config.Provider. The config.Provider
+// is used to provide the initial configuration and update the configuration of
+// the instrumentation in runtime.
+func WithConfigProvider(cp config.Provider) InstrumentationOption {
+	return fnOpt(func(_ context.Context, c instConfig) (instConfig, error) {
+		c.cp = cp
+		return c, nil
+	})
+}

internal/pkg/instrumentation/manager.go

@@ -13,6 +13,9 @@ import (
 	"github.com/cilium/ebpf/rlimit"
 	"github.com/go-logr/logr"
 
+	"go.opentelemetry.io/otel/trace"
+
+	"go.opentelemetry.io/auto/config"
 	dbSql "go.opentelemetry.io/auto/internal/pkg/instrumentation/bpf/database/sql"
 	kafkaConsumer "go.opentelemetry.io/auto/internal/pkg/instrumentation/bpf/github.com/segmentio/kafka-go/consumer"
 	kafkaProducer "go.opentelemetry.io/auto/internal/pkg/instrumentation/bpf/github.com/segmentio/kafka-go/producer"
@@ -35,24 +38,41 @@ var (
 	bpffsCleanup        = bpffs.Cleanup
 )
 
+type managerState int
+
+const (
+	managerStateRunning managerState = iota
+	managerStateStopped
+)
+
 // Manager handles the management of [probe.Probe] instances.
 type Manager struct {
 	logger          logr.Logger
 	probes          map[probe.ID]probe.Probe
 	otelController  *opentelemetry.Controller
 	globalImpl      bool
 	loadedIndicator chan struct{}
+	cp              config.Provider
+	exe             *link.Executable
+	td              *process.TargetDetails
+	runningProbesWG sync.WaitGroup
+	eventCh         chan *probe.Event
+	currentConfig   config.InstrumentationConfig
+	probeMu         sync.Mutex
+	state           managerState
 }
 
 // NewManager returns a new [Manager].
-func NewManager(logger logr.Logger, otelController *opentelemetry.Controller, globalImpl bool, loadIndicator chan struct{}) (*Manager, error) {
+func NewManager(logger logr.Logger, otelController *opentelemetry.Controller, globalImpl bool, loadIndicator chan struct{}, cp config.Provider) (*Manager, error) {
 	logger = logger.WithName("Manager")
 	m := &Manager{
 		logger:          logger,
 		probes:          make(map[probe.ID]probe.Probe),
 		otelController:  otelController,
 		globalImpl:      globalImpl,
 		loadedIndicator: loadIndicator,
+		cp:              cp,
+		eventCh:         make(chan *probe.Event),
 	}
 
 	err := m.registerProbes()
@@ -133,43 +153,147 @@ func (m *Manager) FilterUnusedProbes(target *process.TargetDetails) {
 	}
 }
 
+func getProbeConfig(id probe.ID, c config.InstrumentationConfig) (config.InstrumentationLibrary, bool) {
+	libKindID := config.InstrumentationLibraryID{
+		InstrumentedPkg: id.InstrumentedPkg,
+		SpanKind:        id.SpanKind,
+	}
+
+	if lib, ok := c.InstrumentationLibraryConfigs[libKindID]; ok {
+		return lib, true
+	}
+
+	libID := config.InstrumentationLibraryID{
+		InstrumentedPkg: id.InstrumentedPkg,
+		SpanKind:        trace.SpanKindUnspecified,
+	}
+
+	if lib, ok := c.InstrumentationLibraryConfigs[libID]; ok {
+		return lib, true
+	}
+
+	return config.InstrumentationLibrary{}, false
+}
+
+func isProbeEnabled(id probe.ID, c config.InstrumentationConfig) bool {
+	if pc, ok := getProbeConfig(id, c); ok && pc.TracesEnabled != nil {
+		return *pc.TracesEnabled
+	}
+	return !c.DefaultTracesDisabled
+}
+
+func (m *Manager) applyConfig(c config.InstrumentationConfig) error {
+	if m.td == nil {
+		return errors.New("failed to apply config: target details not set")
+	}
+	if m.exe == nil {
+		return errors.New("failed to apply config: executable not set")
+	}
+
+	var err error
+	m.probeMu.Lock()
+	defer m.probeMu.Unlock()
+
+	if m.state != managerStateRunning {
+		return nil
+	}
+
+	for id, p := range m.probes {
+		currentlyEnabled := isProbeEnabled(id, m.currentConfig)
+		newEnabled := isProbeEnabled(id, c)
+
+		if currentlyEnabled && !newEnabled {
+			m.logger.Info("Disabling probe", "id", id)
+			err = errors.Join(err, p.Close())
+			continue
+		}
+
+		if !currentlyEnabled && newEnabled {
+			m.logger.Info("Enabling probe", "id", id)
+			err = errors.Join(err, p.Load(m.exe, m.td))
+			if err == nil {
+				m.runProbe(p)
+			}
+			continue
+		}
+	}
+
+	return nil
+}
+
+func (m *Manager) runProbe(p probe.Probe) {
+	m.runningProbesWG.Add(1)
+	go func(ap probe.Probe) {
+		defer m.runningProbesWG.Done()
+		ap.Run(m.eventCh)
+	}(p)
+}
+
+func (m *Manager) ConfigLoop(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case c, ok := <-m.cp.Watch():
+			if !ok {
+				m.logger.Info("Configuration provider closed, configuration updates will no longer be received")
+				return
+			}
+			err := m.applyConfig(c)
+			if err != nil {
+				m.logger.Error(err, "Failed to apply config")
+				continue
+			}
+			m.currentConfig = c
+		}
+	}
+}
+
 // Run runs the event processing loop for all managed probes.
 func (m *Manager) Run(ctx context.Context, target *process.TargetDetails) error {
 	if len(m.probes) == 0 {
 		return errors.New("no instrumentation for target process")
 	}
+	if m.cp == nil {
+		return errors.New("no config provider set")
+	}
+
+	m.currentConfig = m.cp.InitialConfig(ctx)
 
 	err := m.load(target)
 	if err != nil {
 		return err
 	}
 
-	eventCh := make(chan *probe.Event)
-	var wg sync.WaitGroup
-	for _, i := range m.probes {
-		wg.Add(1)
-		go func(p probe.Probe) {
-			defer wg.Done()
-			p.Run(eventCh)
-		}(i)
+	for id, p := range m.probes {
+		if isProbeEnabled(id, m.currentConfig) {
+			m.runProbe(p)
+		}
 	}
 
 	if m.loadedIndicator != nil {
 		close(m.loadedIndicator)
 	}
+	m.state = managerStateRunning
+
+	go m.ConfigLoop(ctx)
 
 	for {
 		select {
 		case <-ctx.Done():
+			m.probeMu.Lock()
+
 			m.logger.V(1).Info("Shutting down all probes")
 			err := m.cleanup(target)
 
 			// Wait for all probes to stop before closing the chan they send on.
-			wg.Wait()
-			close(eventCh)
+			m.runningProbesWG.Wait()
+			close(m.eventCh)
 
+			m.state = managerStateStopped
+			m.probeMu.Unlock()
 			return errors.Join(err, ctx.Err())
-		case e := <-eventCh:
+		case e := <-m.eventCh:
 			m.otelController.Trace(e)
 		}
 	}
@@ -185,18 +309,25 @@ func (m *Manager) load(target *process.TargetDetails) error {
 	if err != nil {
 		return err
 	}
+	m.exe = exe
+
+	if m.td == nil {
+		m.td = target
+	}
 
 	if err := m.mount(target); err != nil {
 		return err
 	}
 
 	// Load probes
 	for name, i := range m.probes {
-		m.logger.V(0).Info("loading probe", "name", name)
-		err := i.Load(exe, target)
-		if err != nil {
-			m.logger.Error(err, "error while loading probes, cleaning up", "name", name)
-			return errors.Join(err, m.cleanup(target))
+		if isProbeEnabled(name, m.currentConfig) {
+			m.logger.V(0).Info("loading probe", "name", name)
+			err := i.Load(exe, target)
+			if err != nil {
+				m.logger.Error(err, "error while loading probes, cleaning up", "name", name)
+				return errors.Join(err, m.cleanup(target))
+			}
 		}
 	}
 
@@ -215,6 +346,7 @@ func (m *Manager) mount(target *process.TargetDetails) error {
 
 func (m *Manager) cleanup(target *process.TargetDetails) error {
 	var err error
+	err = errors.Join(err, m.cp.Shutdown(context.Background()))
 	for _, i := range m.probes {
 		err = errors.Join(err, i.Close())
 	}