Verify the language of a process before instrumenting

instrumentation/manager.go

@@ -170,11 +170,7 @@ func (m *manager[ProcessDetails, ConfigGroup]) runEventLoop(ctx context.Context)
 			case detector.ProcessExecEvent:
 				m.logger.V(1).Info("detected new process", "pid", e.PID, "cmd", e.ExecDetails.CmdLine)
 				err := m.handleProcessExecEvent(ctx, e)
-				// ignore the error if no instrumentation factory is found,
-				// as this is expected for some language and sdk combinations
-				if err != nil && !errors.Is(err, errNoInstrumentationFactory) {
-					m.logger.Error(err, "failed to handle process exec event")
-				}
+				m.handleProcessExecEventError(err)
 			case detector.ProcessExitEvent:
 				m.cleanInstrumentation(ctx, e.PID)
 			}
@@ -189,6 +185,27 @@ func (m *manager[ProcessDetails, ConfigGroup]) runEventLoop(ctx context.Context)
 	}
 }
 
+func (m *manager[ProcessDetails, ConfigGroup]) handleProcessExecEventError(err error) {
+	// ignore the error if no instrumentation factory is found,
+	// as this is expected for some language and sdk combinations which don't have ebpf support.
+	if errors.Is(err, errNoInstrumentationFactory) {
+		return
+	}
+
+	// we might fail to get the distribution for the process details,
+	// in cases where we detected a certain language for a container, but multiple processes are running in it,
+	// only one or some of them are in the language we detected. 
+	if errors.Is(err, errFailedToGetDistribution) {
+		m.logger.Info("failed to get otel distribution for process", "error", err)
+		return
+	}
+
+	// fallback to log an error
+	if err != nil {
+		m.logger.Error(err, "failed to handle process exec event")
+	}
+}
+
 func (m *manager[ProcessDetails, ConfigGroup]) Run(ctx context.Context) error {
 	g, errCtx := errgroup.WithContext(ctx)
 
@@ -273,17 +290,23 @@ func (m *manager[ProcessDetails, ConfigGroup]) handleProcessExecEvent(ctx contex
 		// return nil
 	}
 
-	inst, err := factory.CreateInstrumentation(ctx, e.PID, settings)
-	if err != nil {
+	inst, initErr := factory.CreateInstrumentation(ctx, e.PID, settings)
+	reporterErr := m.handler.Reporter.OnInit(ctx, e.PID, err, pd)
+	if reporterErr != nil {
+		m.logger.Error(reporterErr, "failed to report instrumentation init", "initialized", initErr == nil, "pid", e.PID, "process group details", pd)
+	}
+	if initErr != nil {
+		// we need to track the instrumentation even if the initialization failed.
+		// consider a reporter which writes a persistent record for a failed/successful init
+		// we need to notify the reporter once that PID exits to clean up the resources - hence we track it.
+		m.startTrackInstrumentation(e.PID, nil, pd, configGroup)
 		m.logger.Error(err, "failed to initialize instrumentation", "language", otelDisto.Language, "sdk", otelDisto.OtelSdk)
-		err = m.handler.Reporter.OnInit(ctx, e.PID, err, pd)
 		// TODO: should we return here the initialize error? or the handler error? or both?
-		return err
+		return initErr
 	}
 
 	loadErr := inst.Load(ctx)
-
-	reporterErr := m.handler.Reporter.OnLoad(ctx, e.PID, loadErr, pd)
+	reporterErr = m.handler.Reporter.OnLoad(ctx, e.PID, loadErr, pd)
 	if reporterErr != nil {
 		m.logger.Error(reporterErr, "failed to report instrumentation load", "loaded", loadErr == nil, "pid", e.PID, "process group details", pd)
 	}
@@ -295,7 +318,7 @@ func (m *manager[ProcessDetails, ConfigGroup]) handleProcessExecEvent(ctx contex
 		m.startTrackInstrumentation(e.PID, nil, pd, configGroup)
 		m.logger.Error(err, "failed to load instrumentation", "language", otelDisto.Language, "sdk", otelDisto.OtelSdk)
 		// TODO: should we return here the load error? or the instance write error? or both?
-		return err
+		return loadErr
 	}
 
 	m.startTrackInstrumentation(e.PID, inst, pd, configGroup)

odiglet/pkg/ebpf/distribution_matcher.go

@@ -6,18 +6,31 @@ import (
 
 	"github.com/odigos-io/odigos/instrumentation"
 	odgiosK8s "github.com/odigos-io/odigos/k8sutils/pkg/container"
+	"github.com/odigos-io/odigos/procdiscovery/pkg/inspectors"
+	"github.com/odigos-io/odigos/procdiscovery/pkg/process"
 )
 
 type podDeviceDistributionMatcher struct{}
 
 func (dm *podDeviceDistributionMatcher) Distribution(ctx context.Context, e K8sProcessDetails) (instrumentation.OtelDistribution, error) {
 	// get the language and sdk for this process event
 	// based on the pod spec and the container name from the process event
-	// TODO: We should have all the required information in the process event
-	// to determine the language - hence in the future we can improve this
 	lang, sdk, err := odgiosK8s.LanguageSdkFromPodContainer(e.pod, e.containerName)
 	if err != nil {
 		return instrumentation.OtelDistribution{}, fmt.Errorf("failed to get language and sdk: %w", err)
 	}
-	return instrumentation.OtelDistribution{Language: lang, OtelSdk: sdk}, nil
+	// verify the language of the process event
+	if ok := inspectors.VerifyLanguage(process.Details{
+		ProcessID: e.procEvent.PID,
+		ExeName:   e.procEvent.ExecDetails.ExeName,
+		CmdLine:   e.procEvent.ExecDetails.CmdLine,
+		Environments: process.ProcessEnvs{
+			DetailedEnvs:  e.procEvent.ExecDetails.Environments,
+		},
+	}, lang); ok {
+		return instrumentation.OtelDistribution{Language: lang, OtelSdk: sdk}, nil
+	}
+
+	return instrumentation.OtelDistribution{},
+	 	fmt.Errorf("process language does not match the detected language (%s) for container: %s. exe name: %s", lang, e.containerName, e.procEvent.ExecDetails.ExeName)
 }

odiglet/pkg/ebpf/reporter.go

@@ -7,6 +7,7 @@ import (
 	odigosv1 "github.com/odigos-io/odigos/api/odigos/v1alpha1"
 	"github.com/odigos-io/odigos/common"
 	"github.com/odigos-io/odigos/instrumentation"
+	"github.com/odigos-io/odigos/instrumentation/detector"
 	"github.com/odigos-io/odigos/k8sutils/pkg/consts"
 	instance "github.com/odigos-io/odigos/k8sutils/pkg/instrumentation_instance"
 	"github.com/odigos-io/odigos/k8sutils/pkg/workload"
@@ -20,6 +21,7 @@ type K8sProcessDetails struct {
 	pod           *corev1.Pod
 	containerName string
 	pw            *workload.PodWorkload
+	procEvent     detector.ProcessEvent
 }
 
 func (kd K8sProcessDetails) String() string {

odiglet/pkg/ebpf/resolvers.go

@@ -36,6 +36,7 @@ func (dr *k8sDetailsResolver) Resolve(ctx context.Context, event detector.Proces
 		pod:           pod,
 		containerName: containerName,
 		pw:            podWorkload,
+		procEvent:     event,
 	}, nil
 }
 

procdiscovery/pkg/inspectors/langdetect.go

@@ -31,14 +31,14 @@ type VersionInspector interface {
 	GetRuntimeVersion(process *process.Details, containerURL string) *version.Version
 }
 
-var inspectorsList = []LanguageInspector{
-	&golang.GolangInspector{},
-	&java.JavaInspector{},
-	&dotnet.DotnetInspector{},
-	&nodejs.NodejsInspector{},
-	&python.PythonInspector{},
-	&mysql.MySQLInspector{},
-	&nginx.NginxInspector{},
+var inspectorsMap = map[common.ProgrammingLanguage]LanguageInspector{
+	common.GoProgrammingLanguage:         &golang.GolangInspector{},
+	common.JavaProgrammingLanguage:       &java.JavaInspector{},
+	common.DotNetProgrammingLanguage:     &dotnet.DotnetInspector{},
+	common.JavascriptProgrammingLanguage: &nodejs.NodejsInspector{},
+	common.PythonProgrammingLanguage:     &python.PythonInspector{},
+	common.MySQLProgrammingLanguage:      &mysql.MySQLInspector{},
+	common.NginxProgrammingLanguage:      &nginx.NginxInspector{},
 }
 
 // DetectLanguage returns the detected language for the process or
@@ -49,7 +49,7 @@ func DetectLanguage(process process.Details, containerURL string) (common.Progra
 		Language: common.UnknownProgrammingLanguage,
 	}
 
-	for _, i := range inspectorsList {
+	for _, i := range inspectorsMap {
 		languageDetected, detected := i.Inspect(&process)
 		if detected {
 			if detectedProgramLanguageDetails.Language == common.UnknownProgrammingLanguage {
@@ -72,3 +72,13 @@ func DetectLanguage(process process.Details, containerURL string) (common.Progra
 
 	return detectedProgramLanguageDetails, nil
 }
+
+func VerifyLanguage(process process.Details, lang common.ProgrammingLanguage) bool {
+	inspector, ok := inspectorsMap[lang]
+	if !ok {
+		return false
+	}
+
+	_, detected := inspector.Inspect(&process)
+	return detected
+}