Add orchestrator for supporting multiple processes

CHANGELOG.md

@@ -12,6 +12,7 @@ OpenTelemetry Go Automatic Instrumentation adheres to [Semantic Versioning](http
 
 - Add net/http client instrumentor. ([#91](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/91))
 - Add context propagation to net/http server instrumentation. ([#92](https://github.com/open-telemetry/opentelemetry-go-instrumentation/pull/92))
+- Add ability to monitor multiple processes. ([#197](https://github.com/open-telemetry/opentelemetry-go-instrumentation/issues/197))
 
 ### Changed
 

Makefile

@@ -133,6 +133,7 @@ fixtures/%:
 		git clone https://github.com/open-telemetry/opentelemetry-helm-charts.git; \
 	fi
 	helm install test -f .github/workflows/e2e/k8s/collector-helm-values.yml opentelemetry-helm-charts/charts/opentelemetry-collector
+	sleep 5
 	kubectl wait --for=condition=Ready --timeout=60s pod/test-opentelemetry-collector-0
 	kubectl -n default create -f .github/workflows/e2e/k8s/sample-job.yml
 	kubectl wait --for=condition=Complete --timeout=60s job/sample-job

cli/main.go

@@ -15,18 +15,41 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/signal"
+	"runtime"
+	"strings"
 	"syscall"
 
-	"go.opentelemetry.io/auto/pkg/errors"
-	"go.opentelemetry.io/auto/pkg/instrumentors"
+	"google.golang.org/grpc"
+
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+
+	"go.opentelemetry.io/auto"
 	"go.opentelemetry.io/auto/pkg/log"
-	"go.opentelemetry.io/auto/pkg/opentelemetry"
+	"go.opentelemetry.io/auto/pkg/orchestrator"
 	"go.opentelemetry.io/auto/pkg/process"
 )
 
+var (
+	// Controller-local reference to the auto-instrumentation release version.
+	releaseVersion = auto.Version()
+	// Start of this auto-instrumentation's exporter User-Agent header, e.g. ""OTel-Go-Auto-Instrumentation/1.2.3".
+	baseUserAgent = fmt.Sprintf("OTel-Go-Auto-Instrumentation/%s", releaseVersion)
+	// Information about the runtime environment for inclusion in User-Agent, e.g. "go/1.18.2 (linux/amd64)".
+	runtimeInfo = fmt.Sprintf(
+		"%s (%s/%s)",
+		strings.Replace(runtime.Version(), "go", "go/", 1),
+		runtime.GOOS,
+		runtime.GOARCH,
+	)
+	// Combined User-Agent identifying this auto-instrumentation and its runtime environment, see RFC7231 for format considerations.
+	autoinstUserAgent = fmt.Sprintf("%s %s", baseUserAgent, runtimeInfo)
+)
+
 func main() {
 	err := log.Init()
 	if err != nil {
@@ -35,56 +58,49 @@ func main() {
 	}
 
 	log.Logger.V(0).Info("starting Go OpenTelemetry Agent ...")
-	target := process.ParseTargetArgs()
-	if err = target.Validate(); err != nil {
-		log.Logger.Error(err, "invalid target args")
-		return
-	}
-
-	processAnalyzer := process.NewAnalyzer()
-	otelController, err := opentelemetry.NewController()
+	ctx := contextWithSigterm(context.Background())
+	log.Logger.V(0).Info("Establishing connection to OTLP receiver ...")
+	otlpTraceClient := otlptracegrpc.NewClient(
+		otlptracegrpc.WithDialOption(grpc.WithUserAgent(autoinstUserAgent)),
+	)
+	traceExporter, err := otlptrace.New(ctx, otlpTraceClient)
 	if err != nil {
-		log.Logger.Error(err, "unable to create OpenTelemetry controller")
+		log.Logger.Error(err, "unable to connect to OTLP endpoint")
 		return
 	}
-
-	instManager, err := instrumentors.NewManager(otelController)
+	targetArgs := process.ParseTargetArgs()
+	if targetArgs != nil {
+		if err := targetArgs.Validate(); err != nil {
+			log.Logger.Error(err, "invalid target args")
+			return
+		}
+	}
+	r, err := orchestrator.New(ctx, targetArgs, traceExporter)
 	if err != nil {
-		log.Logger.Error(err, "error creating instrumetors manager")
-		return
+		log.Logger.V(0).Error(err, "creating orchestrator")
+	}
+	if err = r.Run(); err != nil {
+		log.Logger.Error(err, "running orchestrator")
 	}
+}
 
-	stopper := make(chan os.Signal, 1)
-	signal.Notify(stopper, os.Interrupt, syscall.SIGTERM)
-	go func() {
-		<-stopper
-		log.Logger.V(0).Info("Got SIGTERM, cleaning up..")
-		processAnalyzer.Close()
-		instManager.Close()
-	}()
+func contextWithSigterm(parent context.Context) context.Context {
+	ctx, cancel := context.WithCancel(parent)
 
-	pid, err := processAnalyzer.DiscoverProcessID(target)
-	if err != nil {
-		if err != errors.ErrInterrupted {
-			log.Logger.Error(err, "error while discovering process id")
-		}
-		return
-	}
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, os.Interrupt, syscall.SIGTERM)
 
-	targetDetails, err := processAnalyzer.Analyze(pid, instManager.GetRelevantFuncs())
-	if err != nil {
-		log.Logger.Error(err, "error while analyzing target process")
-		return
-	}
-	log.Logger.V(0).Info("target process analysis completed", "pid", targetDetails.PID,
-		"go_version", targetDetails.GoVersion, "dependencies", targetDetails.Libraries,
-		"total_functions_found", len(targetDetails.Functions))
+	go func() {
+		defer close(ch)
+		defer signal.Stop(ch)
 
-	instManager.FilterUnusedInstrumentors(targetDetails)
+		select {
+		case <-parent.Done(): // if parent is cancelled, return
+			return
+		case <-ch: // if SIGTERM is received, cancel this context
+			cancel()
+		}
+	}()
 
-	log.Logger.V(0).Info("invoking instrumentors")
-	err = instManager.Run(targetDetails)
-	if err != nil && err != errors.ErrInterrupted {
-		log.Logger.Error(err, "error while running instrumentors")
-	}
+	return ctx
 }

pkg/opentelemetry/controller_test.go → cli/main_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package opentelemetry
+package main
 
 import (
 	"fmt"
@@ -24,5 +24,9 @@ import (
 )
 
 func TestUserAgent(t *testing.T) {
-	assert.Contains(t, autoinstUserAgent, fmt.Sprintf("OTel-Go-Auto-Instrumentation/%s", auto.Version()))
+	assert.Contains(
+		t,
+		autoinstUserAgent,
+		fmt.Sprintf("OTel-Go-Auto-Instrumentation/%s", auto.Version()),
+	)
 }

pkg/instrumentors/manager.go

@@ -30,26 +30,24 @@ import (
 	"go.opentelemetry.io/auto/pkg/process"
 )
 
-var (
-	// Error message returned when unable to find all instrumentation functions.
-	errNotAllFuncsFound = fmt.Errorf("not all functions found for instrumentation")
-)
+// Error message returned when unable to find all instrumentation functions.
+var errNotAllFuncsFound = fmt.Errorf("not all functions found for instrumentation")
 
 // Manager handles the management of [Instrumentor] instances.
 type Manager struct {
 	instrumentors  map[string]Instrumentor
 	done           chan bool
 	incomingEvents chan *events.Event
-	otelController *opentelemetry.Controller
 	allocator      *allocator.Allocator
+	otelController *opentelemetry.Controller
 }
 
 // NewManager returns a new [Manager].
 func NewManager(otelController *opentelemetry.Controller) (*Manager, error) {
 	m := &Manager{
 		instrumentors:  make(map[string]Instrumentor),
 		done:           make(chan bool, 1),
-		incomingEvents: make(chan *events.Event),
+		incomingEvents: make(chan *events.Event, 10),
 		otelController: otelController,
 		allocator:      allocator.New(),
 	}
@@ -84,9 +82,9 @@ func (m *Manager) GetRelevantFuncs() map[string]interface{} {
 	return funcsMap
 }
 
-// FilterUnusedInstrumentors filterers Instrumentors whose functions are
+// filterUnusedInstrumentors filterers Instrumentors whose functions are
 // already instrumented out of the Manager.
-func (m *Manager) FilterUnusedInstrumentors(target *process.TargetDetails) {
+func (m *Manager) filterUnusedInstrumentors(target *process.TargetDetails) {
 	existingFuncMap := make(map[string]interface{})
 	for _, f := range target.Functions {
 		existingFuncMap[f.Name] = nil
@@ -102,7 +100,16 @@ func (m *Manager) FilterUnusedInstrumentors(target *process.TargetDetails) {
 
 		if funcsFound != len(inst.FuncNames()) {
 			if funcsFound > 0 {
-				log.Logger.Error(errNotAllFuncsFound, "some of expected functions not found - check instrumented functions", "instrumentation_name", name, "funcs_found", funcsFound, "funcs_expected", len(inst.FuncNames()))
+				log.Logger.Error(
+					errNotAllFuncsFound,
+					"some of expected functions not found - check instrumented functions",
+					"instrumentation_name",
+					name,
+					"funcs_found",
+					funcsFound,
+					"funcs_expected",
+					len(inst.FuncNames()),
+				)
 			}
 			delete(m.instrumentors, name)
 		}

pkg/instrumentors/runner.go

@@ -28,16 +28,18 @@ import (
 
 // Run runs the event processing loop for all managed Instrumentors.
 func (m *Manager) Run(target *process.TargetDetails) error {
+	m.filterUnusedInstrumentors(target)
+
 	if len(m.instrumentors) == 0 {
 		log.Logger.V(0).Info("there are no available instrumentations for target process")
 		return nil
 	}
+	log.Logger.V(0).Info("running instrumentors", "number", fmt.Sprintf("%d", len(m.instrumentors)))
 
 	err := m.load(target)
 	if err != nil {
 		return err
 	}
-
 	for _, i := range m.instrumentors {
 		go i.Run(m.incomingEvents)
 	}
@@ -91,12 +93,12 @@ func (m *Manager) load(target *process.TargetDetails) error {
 		}
 	}
 
-	log.Logger.V(0).Info("loaded instrumentors to memory", "total_instrumentors", len(m.instrumentors))
+	log.Logger.V(0).
+		Info("loaded instrumentors to memory", "total_instrumentors", len(m.instrumentors))
 	return nil
 }
 
 func (m *Manager) cleanup() {
-	close(m.incomingEvents)
 	for _, i := range m.instrumentors {
 		i.Close()
 	}

pkg/opentelemetry/controller.go

@@ -16,39 +16,19 @@ package opentelemetry
 
 import (
 	"context"
-	"fmt"
-	"os"
 	"runtime"
-	"strings"
 	"time"
 
 	"golang.org/x/sys/unix"
-	"google.golang.org/grpc"
 
-	"go.opentelemetry.io/auto"
-	"go.opentelemetry.io/auto/pkg/instrumentors/events"
-	"go.opentelemetry.io/auto/pkg/log"
-	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
-	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
 	"go.opentelemetry.io/otel/trace"
-)
 
-const (
-	otelServiceNameEnvVar = "OTEL_SERVICE_NAME"
-)
-
-var (
-	// Controller-local reference to the auto-instrumentation release version.
-	releaseVersion = auto.Version()
-	// Start of this auto-instrumentation's exporter User-Agent header, e.g. ""OTel-Go-Auto-Instrumentation/1.2.3".
-	baseUserAgent = fmt.Sprintf("OTel-Go-Auto-Instrumentation/%s", releaseVersion)
-	// Information about the runtime environment for inclusion in User-Agent, e.g. "go/1.18.2 (linux/amd64)".
-	runtimeInfo = fmt.Sprintf("%s (%s/%s)", strings.Replace(runtime.Version(), "go", "go/", 1), runtime.GOOS, runtime.GOARCH)
-	// Combined User-Agent identifying this auto-instrumentation and its runtime environment, see RFC7231 for format considerations.
-	autoinstUserAgent = fmt.Sprintf("%s %s", baseUserAgent, runtimeInfo)
+	"go.opentelemetry.io/auto"
+	"go.opentelemetry.io/auto/pkg/instrumentors/events"
+	"go.opentelemetry.io/auto/pkg/log"
 )
 
 // Controller handles OpenTelemetry telemetry generation for events.
@@ -76,7 +56,6 @@ func (c *Controller) Trace(event *events.Event) {
 
 	if event.SpanContext == nil {
 		log.Logger.V(0).Info("got event without context - dropping")
-		return
 	}
 
 	// TODO: handle remote parent
@@ -98,35 +77,23 @@ func (c *Controller) convertTime(t int64) time.Time {
 }
 
 // NewController returns a new initialized [Controller].
-func NewController() (*Controller, error) {
-	serviceName, exists := os.LookupEnv(otelServiceNameEnvVar)
-	if !exists {
-		return nil, fmt.Errorf("%s env var must be set", otelServiceNameEnvVar)
-	}
-
-	ctx := context.Background()
+func NewController(
+	ctx context.Context,
+	serviceName string,
+	exporter sdktrace.SpanExporter,
+) (*Controller, error) {
 	res, err := resource.New(ctx,
 		resource.WithAttributes(
 			semconv.ServiceNameKey.String(serviceName),
 			semconv.TelemetrySDKLanguageGo,
-			semconv.TelemetryAutoVersionKey.String(releaseVersion),
+			semconv.TelemetryAutoVersionKey.String(auto.Version()),
 		),
 	)
 	if err != nil {
 		return nil, err
 	}
 
-	log.Logger.V(0).Info("Establishing connection to OTLP receiver ...")
-	otlpTraceClient := otlptracegrpc.NewClient(
-		otlptracegrpc.WithDialOption(grpc.WithUserAgent(autoinstUserAgent)),
-	)
-	traceExporter, err := otlptrace.New(ctx, otlpTraceClient)
-	if err != nil {
-		log.Logger.Error(err, "unable to connect to OTLP endpoint")
-		return nil, err
-	}
-
-	bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
+	bsp := sdktrace.NewBatchSpanProcessor(exporter)
 	tracerProvider := sdktrace.NewTracerProvider(
 		sdktrace.WithSampler(sdktrace.AlwaysSample()),
 		sdktrace.WithResource(res),