Add programming language selector (#1326)

Author: grcevski

internal/test/integration/configs/obi-config-multiexec-lang.yml

@@ -0,0 +1,23 @@
+log_config: yaml
+routes:
+  patterns:
+    - /basic/:rnd
+  unmatched: path
+otel_metrics_export:
+  endpoint: http://otelcol:4018
+otel_traces_export:
+  endpoint: http://jaeger:4318
+discovery:
+  instrument:
+    - languages: "{rust,ruby}"
+  exclude_instrument:
+    - exe_path: "{obi,prometheus,otelcol*,all*,launcher}"
+attributes:
+  kubernetes:
+    enable: true
+    cluster_name: my-kube
+  select:
+    http_server_request_duration_seconds_count:
+      exclude: ["server_address"]
+    "*":
+      include: ["*"]

internal/test/integration/docker-compose-multiexec.yml

@@ -120,7 +120,7 @@ services:
       context: ../../..
       dockerfile: ./internal/test/integration/components/obi/Dockerfile
     command:
-      - --config=/configs/obi-config-multiexec.yml
+      - --config=/configs/obi-config-multiexec${MULTI_TEST_MODE}.yml
     volumes:
       - ./configs/:/configs
       - ./system/sys/kernel/security:/sys/kernel/security

internal/test/integration/multiprocess_test.go

@@ -4,6 +4,7 @@
 package integration
 
 import (
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"path"
@@ -15,7 +16,9 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/obi/internal/test/integration/components/docker"
+	"go.opentelemetry.io/obi/internal/test/integration/components/jaeger"
 	"go.opentelemetry.io/obi/internal/test/integration/components/promtest"
+	ti "go.opentelemetry.io/obi/pkg/test/integration"
 )
 
 func TestMultiProcess(t *testing.T) {
@@ -226,3 +229,114 @@ func checkInstrumentedProcessesMetric(t *testing.T) {
 		}
 	}, testTimeout, 1000*time.Millisecond)
 }
+
+// We are instrumenting only the Rust and Ruby services, all other server span queries should come empty
+func testPartialLanguageHTTPProbes(t *testing.T) {
+	waitForTestComponentsSub(t, "http://localhost:8091", "/dist") // rust
+
+	for i := 0; i < 100; i++ {
+		ti.DoHTTPGet(t, "http://localhost:8091/dist", 200)
+	}
+
+	// check the rust service, it will not have any nested spans
+	require.EventuallyWithT(t, func(ct *assert.CollectT) {
+		resp, err := http.Get(jaegerQueryURL + "?service=greetings&operation=GET%20%2Fdist")
+		require.NoError(ct, err)
+		if resp == nil {
+			return
+		}
+		require.Equal(ct, http.StatusOK, resp.StatusCode)
+		var tq jaeger.TracesQuery
+		require.NoError(ct, json.NewDecoder(resp.Body).Decode(&tq))
+		traces := tq.FindBySpan(jaeger.Tag{Key: "url.path", Type: "string", Value: "/dist"})
+		require.LessOrEqual(ct, 5, len(traces))
+		for _, trace := range traces {
+			// Check the information of the rust parent span
+			res := trace.FindByOperationName("GET /dist", "server")
+			require.Len(ct, res, 1)
+			parent := res[0]
+			require.NotEmpty(ct, parent.TraceID)
+			require.NotEmpty(ct, parent.SpanID)
+			// check duration is at least 2us
+			assert.Less(ct, (2 * time.Microsecond).Microseconds(), parent.Duration)
+			// check span attributes
+			sd := parent.Diff(
+				jaeger.Tag{Key: "http.request.method", Type: "string", Value: "GET"},
+				jaeger.Tag{Key: "http.response.status_code", Type: "int64", Value: float64(200)},
+				jaeger.Tag{Key: "url.path", Type: "string", Value: "/dist"},
+				jaeger.Tag{Key: "server.port", Type: "int64", Value: float64(8090)},
+				jaeger.Tag{Key: "http.route", Type: "string", Value: "/dist"},
+				jaeger.Tag{Key: "span.kind", Type: "string", Value: "server"},
+			)
+			assert.Empty(ct, sd, sd.String())
+
+			// Check the information of the java parent span
+			res = trace.FindByOperationName("GET /jtrace", "server")
+			require.Empty(ct, res)
+
+			// Check the information of the nodejs parent span
+			res = trace.FindByOperationName("GET /traceme", "server")
+			require.Empty(ct, res)
+
+			// Check the information of the go parent span
+			res = trace.FindByOperationName("GET /gotracemetoo", "server")
+			require.Empty(ct, res)
+
+			// Check the information of the python parent span
+			res = trace.FindByOperationName("GET /tracemetoo", "server")
+			require.Empty(t, res)
+
+			// Check the information of the rails parent span
+			res = trace.FindByOperationName("GET /users", "server")
+			require.Empty(t, res)
+		}
+	}, testTimeout, 100*time.Millisecond)
+
+	require.EventuallyWithT(t, func(ct *assert.CollectT) {
+		resp, err := http.Get(jaegerQueryURL + "?service=ruby&operation=GET%20%2Fusers")
+		require.NoError(ct, err)
+		if resp == nil {
+			return
+		}
+		require.Equal(ct, http.StatusOK, resp.StatusCode)
+		var tq jaeger.TracesQuery
+		require.NoError(ct, json.NewDecoder(resp.Body).Decode(&tq))
+		traces := tq.FindBySpan(jaeger.Tag{Key: "url.path", Type: "string", Value: "/users"})
+		require.LessOrEqual(ct, 5, len(traces))
+		for _, trace := range traces {
+			// Check the information of the rust parent span
+			res := trace.FindByOperationName("GET /users", "server")
+			require.Len(ct, res, 1)
+		}
+	}, testTimeout, 100*time.Millisecond)
+
+	require.EventuallyWithT(t, func(ct *assert.CollectT) {
+		resp, err := http.Get(jaegerQueryURL + "?service=testserver&operation=GET%20%2Fgotracemetoo")
+		require.NoError(ct, err)
+		if resp == nil {
+			return
+		}
+		require.Equal(ct, http.StatusOK, resp.StatusCode)
+		var tq jaeger.TracesQuery
+		require.NoError(ct, json.NewDecoder(resp.Body).Decode(&tq))
+		traces := tq.FindBySpan(jaeger.Tag{Key: "url.path", Type: "string", Value: "/gotracemetoo"})
+		require.Empty(ct, traces)
+	}, testTimeout, 100*time.Millisecond)
+}
+
+func TestLanguageSelectors(t *testing.T) {
+	compose, err := docker.ComposeSuite("docker-compose-multiexec.yml", path.Join(pathOutput, "test-suite-multiexec-lang.log"))
+	require.NoError(t, err)
+
+	// we are going to setup discovery directly in the configuration file, choose the lang config file
+	compose.Env = append(compose.Env, `OTEL_EBPF_EXECUTABLE_PATH=`, `OTEL_EBPF_OPEN_PORT=`, `MULTI_TEST_MODE=-lang`)
+	require.NoError(t, compose.Up())
+
+	// We are testing with instrumenting only Ruby and Rust services, so from our call chain we should only see
+	// traces for the two services written in the correct language
+	t.Run("Partial traces: rust (OK) -> java (NO) -> node (NO) -> go (NO) -> python (NO) -> rails (OK)", func(t *testing.T) {
+		testPartialLanguageHTTPProbes(t)
+	})
+
+	require.NoError(t, compose.Close())
+}

pkg/appolly/discover/language_decorator.go

@@ -74,6 +74,7 @@ func (ld *languageDecorator) decorateEvent(ev *Event[ProcessAttrs]) {
 			}
 			t := _findProcLanguage(ev.Obj.pid)
 			ev.Obj.detectedType = t
+			ld.log.Debug("detected type", "pid", ev.Obj.pid, "type", t)
 			ld.typeCache.Add(ino, t)
 		}
 	}

pkg/appolly/discover/language_decorator_test.go

@@ -4,6 +4,7 @@
 package discover
 
 import (
+	"log/slog"
 	"testing"
 
 	lru "github.com/hashicorp/golang-lru/v2"
@@ -17,6 +18,7 @@ func newTestDecorator(ignoredPaths []string) *languageDecorator {
 	cache, _ := lru.New[uint64, svc.InstrumentableType](100)
 	return &languageDecorator{
 		typeCache:    cache,
+		log:          slog.With("component", "LanguageDecorator"),
 		ignoredPaths: ignoredPaths,
 	}
 }

pkg/appolly/discover/matcher.go

@@ -211,7 +211,7 @@ func (m *Matcher) isExcluded(obj *ProcessAttrs, proc *services.ProcessInfo) bool
 
 func (m *Matcher) matchProcess(obj *ProcessAttrs, p *services.ProcessInfo, a services.Selector) bool {
 	log := m.Log.With("pid", p.Pid, "exe", p.ExePath)
-	if !a.GetPath().IsSet() && a.GetOpenPorts().Len() == 0 && len(obj.metadata) == 0 {
+	if !a.GetPath().IsSet() && !a.GetLanguages().IsSet() && a.GetOpenPorts().Len() == 0 && len(obj.metadata) == 0 {
 		log.Debug("no Kube metadata, no local selection criteria. Ignoring")
 		return false
 	}
@@ -223,6 +223,10 @@ func (m *Matcher) matchProcess(obj *ProcessAttrs, p *services.ProcessInfo, a ser
 		log.Debug("open ports do not match", "openPorts", a.GetOpenPorts(), "process ports", p.OpenPorts)
 		return false
 	}
+	if a.GetLanguages().IsSet() && !m.matchByLanguage(obj, a) {
+		log.Debug("executable language does not match", "languages", a.GetLanguages(), "type", obj.detectedType.String())
+		return false
+	}
 	if a.IsContainersOnly() {
 		ns, _ := namespaceFetcherFunc(p.Pid)
 		if ns == m.Namespace && m.HasHostPidAccess {
@@ -253,6 +257,10 @@ func (m *Matcher) matchByExecutable(p *services.ProcessInfo, a services.Selector
 	return a.GetPathRegexp().MatchString(p.ExePath)
 }
 
+func (m *Matcher) matchByLanguage(actual *ProcessAttrs, a services.Selector) bool {
+	return a.GetLanguages().MatchString(actual.detectedType.String())
+}
+
 func (m *Matcher) matchByAttributes(actual *ProcessAttrs, required services.Selector) bool {
 	if required == nil {
 		return true

pkg/appolly/discover/matcher_test.go

@@ -11,6 +11,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"go.opentelemetry.io/obi/pkg/appolly/app"
+	"go.opentelemetry.io/obi/pkg/appolly/app/svc"
 	"go.opentelemetry.io/obi/pkg/appolly/services"
 	"go.opentelemetry.io/obi/pkg/internal/testutil"
 	"go.opentelemetry.io/obi/pkg/obi"
@@ -76,6 +77,51 @@ func TestCriteriaMatcher(t *testing.T) {
 	testMatch(t, matches[3], "exec-only", "", services.ProcessInfo{Pid: 6, ExePath: "/bin/clientweird99"})
 }
 
+func TestCriteriaMatcherLanguage(t *testing.T) {
+	pipeConfig := obi.Config{}
+	require.NoError(t, yaml.Unmarshal([]byte(`discovery:
+  services:
+  - name: go-and-java
+    namespace: foo
+    languages: "go|java"
+  - name: rust
+    languages: rust
+`), &pipeConfig))
+
+	discoveredProcesses := msg.NewQueue[[]Event[ProcessAttrs]](msg.ChannelBufferLen(10))
+	filteredProcessesQu := msg.NewQueue[[]Event[ProcessMatch]](msg.ChannelBufferLen(10))
+	filteredProcesses := filteredProcessesQu.Subscribe()
+	matcherFunc, err := criteriaMatcherProvider(&pipeConfig, discoveredProcesses, filteredProcessesQu)(t.Context())
+	require.NoError(t, err)
+	go matcherFunc(t.Context())
+	defer filteredProcessesQu.Close()
+
+	// it will filter unmatching processes and return a ProcessMatch for these that match
+	processInfo = func(pp ProcessAttrs) (*services.ProcessInfo, error) {
+		exePath := map[app.PID]string{
+			1: "/bin/weird33", 2: "/bin/weird33", 3: "server",
+			4: "/bin/something", 5: "server", 6: "/bin/clientweird99",
+		}[pp.pid]
+		return &services.ProcessInfo{Pid: pp.pid, ExePath: exePath, OpenPorts: pp.openPorts}, nil
+	}
+	discoveredProcesses.Send([]Event[ProcessAttrs]{
+		{Type: EventCreated, Obj: ProcessAttrs{pid: 1, openPorts: []uint32{1, 2, 3}, detectedType: svc.InstrumentableCPP}},     // filter
+		{Type: EventDeleted, Obj: ProcessAttrs{pid: 2, openPorts: []uint32{4}, detectedType: svc.InstrumentableGeneric}},       // filter
+		{Type: EventCreated, Obj: ProcessAttrs{pid: 3, openPorts: []uint32{8433}, detectedType: svc.InstrumentableJavaNative}}, // pass
+		{Type: EventCreated, Obj: ProcessAttrs{pid: 4, openPorts: []uint32{8083}, detectedType: svc.InstrumentableJava}},       // pass
+		{Type: EventCreated, Obj: ProcessAttrs{pid: 5, openPorts: []uint32{443}, detectedType: svc.InstrumentableGolang}},      // pass
+		{Type: EventCreated, Obj: ProcessAttrs{pid: 6, detectedType: svc.InstrumentableRust}},                                  // pass
+	})
+
+	matches := testutil.ReadChannel(t, filteredProcesses, testTimeout)
+	require.Len(t, matches, 4)
+
+	testMatch(t, matches[0], "go-and-java", "foo", services.ProcessInfo{Pid: 3, ExePath: "server", OpenPorts: []uint32{8433}})
+	testMatch(t, matches[1], "go-and-java", "foo", services.ProcessInfo{Pid: 4, ExePath: "/bin/something", OpenPorts: []uint32{8083}})
+	testMatch(t, matches[2], "go-and-java", "foo", services.ProcessInfo{Pid: 5, ExePath: "server", OpenPorts: []uint32{443}})
+	testMatch(t, matches[3], "rust", "", services.ProcessInfo{Pid: 6, ExePath: "/bin/clientweird99"})
+}
+
 func TestCriteriaMatcher_Exclude(t *testing.T) {
 	pipeConfig := obi.Config{}
 	require.NoError(t, yaml.Unmarshal([]byte(`discovery:

pkg/appolly/discover/typer_test.go

@@ -28,6 +28,7 @@ type dummyCriterion struct {
 func (d dummyCriterion) GetName() string                                                { return d.name }
 func (d dummyCriterion) GetOpenPorts() *services.PortEnum                               { return nil }
 func (d dummyCriterion) GetPath() services.StringMatcher                                { return nil }
+func (d dummyCriterion) GetLanguages() services.StringMatcher                           { return nil }
 func (d dummyCriterion) RangeMetadata() iter.Seq2[string, services.StringMatcher]       { return nil }
 func (d dummyCriterion) RangePodAnnotations() iter.Seq2[string, services.StringMatcher] { return nil }
 func (d dummyCriterion) RangePodLabels() iter.Seq2[string, services.StringMatcher]      { return nil }

pkg/appolly/services/attr_glob.go

@@ -23,6 +23,7 @@ func (dc GlobDefinitionCriteria) Validate() error {
 	for i := range dc {
 		if dc[i].OpenPorts.Len() == 0 &&
 			!dc[i].Path.IsSet() &&
+			!dc[i].Languages.IsSet() &&
 			len(dc[i].Metadata) == 0 &&
 			len(dc[i].PodLabels) == 0 &&
 			len(dc[i].PodAnnotations) == 0 {
@@ -61,6 +62,10 @@ type GlobAttributes struct {
 	// list of port numbers (e.g. 80) and port ranges (e.g. 8080-8089)
 	OpenPorts PortEnum `yaml:"open_ports"`
 
+	// Language allows defining services to instrument based on the
+	// programming language they are written in. Use lowercase names, e.g. java,go
+	Languages GlobAttr `yaml:"languages"`
+
 	// Path allows defining the regular expression matching the full executable path.
 	Path GlobAttr `yaml:"exe_path"`
 
@@ -150,6 +155,7 @@ func (p *GlobAttr) MatchString(input string) bool {
 func (ga *GlobAttributes) GetName() string                        { return ga.Name }
 func (ga *GlobAttributes) GetNamespace() string                   { return ga.Namespace }
 func (ga *GlobAttributes) GetPath() StringMatcher                 { return &ga.Path }
+func (ga *GlobAttributes) GetLanguages() StringMatcher            { return &ga.Languages }
 func (ga *GlobAttributes) GetPathRegexp() StringMatcher           { return nilMatcher{} }
 func (ga *GlobAttributes) GetOpenPorts() *PortEnum                { return &ga.OpenPorts }
 func (ga *GlobAttributes) IsContainersOnly() bool                 { return ga.ContainersOnly }

pkg/appolly/services/attr_regex.go

@@ -24,6 +24,7 @@ func (dc RegexDefinitionCriteria) Validate() error {
 		if dc[i].OpenPorts.Len() == 0 &&
 			!dc[i].Path.IsSet() &&
 			!dc[i].PathRegexp.IsSet() &&
+			!dc[i].Languages.IsSet() &&
 			len(dc[i].Metadata) == 0 &&
 			len(dc[i].PodLabels) == 0 &&
 			len(dc[i].PodAnnotations) == 0 {
@@ -66,6 +67,9 @@ type RegexSelector struct {
 	OpenPorts PortEnum `yaml:"open_ports"`
 	// Path allows defining the regular expression matching the full executable path.
 	Path RegexpAttr `yaml:"exe_path"`
+	// Language allows defining services to instrument based on the
+	// programming language they are written in.
+	Languages RegexpAttr `yaml:"languages"`
 	// PathRegexp is deprecated but kept here for backwards compatibility with Beyla 1.0.x.
 	// Deprecated. Please use Path (exe_path YAML attribute)
 	PathRegexp RegexpAttr `yaml:"exe_path_regexp"`
@@ -155,6 +159,7 @@ func (p *RegexpAttr) MatchString(input string) bool {
 func (a *RegexSelector) GetName() string                        { return a.Name }
 func (a *RegexSelector) GetNamespace() string                   { return a.Namespace }
 func (a *RegexSelector) GetPath() StringMatcher                 { return &a.Path }
+func (a *RegexSelector) GetLanguages() StringMatcher            { return &a.Languages }
 func (a *RegexSelector) GetPathRegexp() StringMatcher           { return &a.PathRegexp }
 func (a *RegexSelector) GetOpenPorts() *PortEnum                { return &a.OpenPorts }
 func (a *RegexSelector) IsContainersOnly() bool                 { return a.ContainersOnly }