feat: use annotation for ebpf instrumentation (#766)

common/consts/consts.go

@@ -13,6 +13,7 @@ const (
 	InstrumentationDisabled        = "disabled"
 	GolangInstrumentationImage     = "keyval/otel-go-agent:v0.6.5"
 	OdigosReportedNameAnnotation   = "odigos.io/reported-name"
+	EbpfInstrumentationAnnotation  = "instrumentation.odigos.io/ebpf"
 )
 
 var (

common/device_names.go

@@ -5,15 +5,13 @@ type OdigosInstrumentationDevice string
 const (
 	JavaDeviceName       OdigosInstrumentationDevice = "instrumentation.odigos.io/java"
 	PythonDeviceName     OdigosInstrumentationDevice = "instrumentation.odigos.io/python"
-	GoDeviceName         OdigosInstrumentationDevice = "instrumentation.odigos.io/go"
 	DotNetDeviceName     OdigosInstrumentationDevice = "instrumentation.odigos.io/dotnet"
 	JavascriptDeviceName OdigosInstrumentationDevice = "instrumentation.odigos.io/javascript"
 )
 
 var InstrumentationDevices = []OdigosInstrumentationDevice{
 	JavaDeviceName,
 	PythonDeviceName,
-	GoDeviceName,
 	DotNetDeviceName,
 	JavascriptDeviceName,
 }
@@ -24,8 +22,6 @@ func ProgrammingLanguageToInstrumentationDevice(language ProgrammingLanguage) Od
 		return JavaDeviceName
 	case PythonProgrammingLanguage:
 		return PythonDeviceName
-	case GoProgrammingLanguage:
-		return GoDeviceName
 	case DotNetProgrammingLanguage:
 		return DotNetDeviceName
 	case JavascriptProgrammingLanguage:

instrumentor/controllers/common.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/go-logr/logr"
 	odigosv1 "github.com/keyval-dev/odigos/api/odigos/v1alpha1"
+	"github.com/keyval-dev/odigos/common"
 	"github.com/keyval-dev/odigos/common/consts"
 	"github.com/keyval-dev/odigos/common/utils"
 	"github.com/keyval-dev/odigos/instrumentor/instrumentation"
@@ -25,6 +26,36 @@ var (
 	IgnoredNamespaces map[string]bool
 )
 
+// shouldInstrumentWithEbpf returns true if the given runtime details should be delegated to odiglet for ebpf instrumentation
+// This is currently hardcoded. In the future we will read this from a config
+func shouldInstrumentWithEbpf(runtimeDetails *odigosv1.InstrumentedApplication) bool {
+	for _, l := range runtimeDetails.Spec.Languages {
+		if l.Language == common.GoProgrammingLanguage {
+			return true
+		}
+	}
+
+	return false
+}
+
+func setInstrumentationEbpf(obj client.Object) {
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		annotations = make(map[string]string)
+	}
+
+	annotations[consts.EbpfInstrumentationAnnotation] = "true"
+}
+
+func clearInstrumentationEbpf(obj client.Object) {
+	annotations := obj.GetAnnotations()
+	if annotations == nil {
+		return
+	}
+
+	delete(annotations, consts.EbpfInstrumentationAnnotation)
+}
+
 func isDataCollectionReady(ctx context.Context, c client.Client) bool {
 	logger := log.FromContext(ctx)
 	var collectorGroups odigosv1.CollectorsGroupList
@@ -50,6 +81,11 @@ func instrument(logger logr.Logger, ctx context.Context, kubeClient client.Clien
 	}
 
 	result, err := controllerutil.CreateOrPatch(ctx, kubeClient, obj, func() error {
+		if shouldInstrumentWithEbpf(runtimeDetails) {
+			setInstrumentationEbpf(obj)
+			return nil
+		}
+
 		podSpec, err := getPodSpecFromObject(obj)
 		if err != nil {
 			return err
@@ -90,6 +126,7 @@ func uninstrument(logger logr.Logger, ctx context.Context, kubeClient client.Cli
 	}
 
 	result, err := controllerutil.CreateOrPatch(ctx, kubeClient, obj, func() error {
+		clearInstrumentationEbpf(obj)
 		podSpec, err := getPodSpecFromObject(obj)
 		if err != nil {
 			return err

odiglet/cmd/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kubevirt/device-plugin-manager/pkg/dpm"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
 )
 
 func main() {
@@ -47,9 +48,11 @@ func main() {
 
 	go startDeviceManager(clientset)
 
-	ctx, err := kube.StartReconciling(ebpfDirectors)
+	ctx := signals.SetupSignalHandler()
+
+	err = kube.StartReconciling(ctx, ebpfDirectors)
 	if err != nil {
-		log.Logger.Error(err, "Failed to start reconciling")
+		log.Logger.Error(err, "Failed to start controller-runtime manager")
 		os.Exit(-1)
 	}
 

odiglet/pkg/ebpf/director.go

@@ -2,14 +2,11 @@ package ebpf
 
 import (
 	"context"
-	"errors"
 
 	"github.com/keyval-dev/odigos/common"
 	"k8s.io/apimachinery/pkg/types"
 )
 
-var ErrProcInstrumented = errors.New("process already instrumented")
-
 type Director interface {
 	Language() common.ProgrammingLanguage
 	Instrument(ctx context.Context, pid int, podDetails types.NamespacedName, appName string) error

odiglet/pkg/ebpf/go.go

@@ -15,15 +15,26 @@ import (
 )
 
 type InstrumentationDirectorGo struct {
-	mux                   sync.Mutex
+	mux sync.Mutex
+
+	// this map holds the instrumentation object which is used to close the instrumentation
+	// the map is filled only after the instrumentation is actually created
+	// which is an asyn process that might take some time
 	pidsToInstrumentation map[int]*auto.Instrumentation
-	podDetailsToPids      map[types.NamespacedName][]int
+
+	// this map is used to make sure we do not attempt to instrument the same process twice.
+	// it keeps track of which processes we already attempted to instrument,
+	// so we can avoid attempting to instrument them again.
+	pidsAttemptedInstrumentation map[int]struct{}
+
+	podDetailsToPids map[types.NamespacedName][]int
 }
 
 func NewInstrumentationDirectorGo() (Director, error) {
 	return &InstrumentationDirectorGo{
-		pidsToInstrumentation: make(map[int]*auto.Instrumentation),
-		podDetailsToPids:      make(map[types.NamespacedName][]int),
+		pidsToInstrumentation:        make(map[int]*auto.Instrumentation),
+		pidsAttemptedInstrumentation: make(map[int]struct{}),
+		podDetailsToPids:             make(map[types.NamespacedName][]int),
 	}, nil
 }
 
@@ -35,10 +46,12 @@ func (i *InstrumentationDirectorGo) Instrument(ctx context.Context, pid int, pod
 	log.Logger.V(0).Info("Instrumenting process", "pid", pid)
 	i.mux.Lock()
 	defer i.mux.Unlock()
-	if _, exists := i.pidsToInstrumentation[pid]; exists {
+	if _, exists := i.pidsAttemptedInstrumentation[pid]; exists {
 		log.Logger.V(5).Info("Process already instrumented", "pid", pid)
-		return ErrProcInstrumented
+		return nil
 	}
+	i.podDetailsToPids[podDetails] = append(i.podDetailsToPids[podDetails], pid)
+	i.pidsAttemptedInstrumentation[pid] = struct{}{}
 
 	defaultExporter, err := otlptracegrpc.New(
 		ctx,
@@ -62,8 +75,21 @@ func (i *InstrumentationDirectorGo) Instrument(ctx context.Context, pid int, pod
 			return
 		}
 
-		i.pidsToInstrumentation[pid] = inst
-		i.podDetailsToPids[podDetails] = append(i.podDetailsToPids[podDetails], pid)
+		i.mux.Lock()
+		_, stillExists := i.pidsAttemptedInstrumentation[pid]
+		if stillExists {
+			i.pidsToInstrumentation[pid] = inst
+			i.mux.Unlock()
+		} else {
+			i.mux.Unlock()
+			// we attempted to instrument this process, but it was already cleaned up
+			// so we need to clean up the instrumentation we just created
+			err = inst.Close()
+			if err != nil {
+				log.Logger.Error(err, "error cleaning up instrumentation for process", "pid", pid)
+			}
+			return
+		}
 
 		if err := inst.Run(context.Background()); err != nil {
 			log.Logger.Error(err, "instrumentation crashed after running")
@@ -82,9 +108,11 @@ func (i *InstrumentationDirectorGo) Cleanup(podDetails types.NamespacedName) {
 		return
 	}
 
-	log.Logger.V(0).Info("Cleaning up instrumentation for pod", "pod", podDetails)
+	log.Logger.V(0).Info("Cleaning up ebpf go instrumentation for pod", "pod", podDetails)
 	delete(i.podDetailsToPids, podDetails)
 	for _, pid := range pids {
+		delete(i.pidsAttemptedInstrumentation, pid)
+
 		inst, exists := i.pidsToInstrumentation[pid]
 		if !exists {
 			log.Logger.V(5).Info("No objects to cleanup for process", "pid", pid)

odiglet/pkg/kube/instrumentation_ebpf/daemonsets.go

@@ -0,0 +1,27 @@
+package instrumentation_ebpf
+
+import (
+	"context"
+
+	"github.com/keyval-dev/odigos/common"
+	"github.com/keyval-dev/odigos/odiglet/pkg/ebpf"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type DaemonSetsReconciler struct {
+	client.Client
+	Scheme    *runtime.Scheme
+	Directors map[common.ProgrammingLanguage]ebpf.Director
+}
+
+func (d *DaemonSetsReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) {
+	err := ApplyEbpfToPodWorkload(ctx, d.Client, d.Directors, &PodWorkload{
+		Name:      request.Name,
+		Namespace: request.Namespace,
+		Kind:      "DaemonSet",
+	})
+
+	return ctrl.Result{}, err
+}

odiglet/pkg/kube/instrumentation_ebpf/deployments.go

@@ -0,0 +1,27 @@
+package instrumentation_ebpf
+
+import (
+	"context"
+
+	"github.com/keyval-dev/odigos/common"
+	"github.com/keyval-dev/odigos/odiglet/pkg/ebpf"
+	"k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type DeploymentsReconciler struct {
+	client.Client
+	Scheme    *runtime.Scheme
+	Directors map[common.ProgrammingLanguage]ebpf.Director
+}
+
+func (d *DeploymentsReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) {
+	err := ApplyEbpfToPodWorkload(ctx, d.Client, d.Directors, &PodWorkload{
+		Name:      request.Name,
+		Namespace: request.Namespace,
+		Kind:      "Deployment",
+	})
+
+	return ctrl.Result{}, err
+}