Enable Config v2 in the OBI receiver (#2681)

Author: MrAlias

.github/workflows/pull_request.yml

@@ -241,3 +241,11 @@ jobs:
       - name: Build collector distribution
         working-directory: examples/otel-collector
         run: "$(go env GOPATH)/bin/builder --config ./builder-config.yaml"
+
+      - name: Validate Config v2 collector configuration
+        working-directory: examples/otel-collector
+        run: ./otelcol-dev/otelcol-dev validate --config ./config.yaml
+
+      - name: Smoke test Config v2 collector receiver
+        working-directory: examples/otel-collector
+        run: sudo ./smoke-test.sh

collector/config_linux.go

@@ -0,0 +1,76 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build linux && (amd64 || arm64)
+
+package collector // import "go.opentelemetry.io/obi/collector"
+
+import (
+	"errors"
+	"fmt"
+
+	"go.yaml.in/yaml/v3"
+
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+
+	"go.opentelemetry.io/obi/internal/config/convert"
+	"go.opentelemetry.io/obi/internal/config/schema"
+	"go.opentelemetry.io/obi/pkg/obi"
+)
+
+type receiverConfig struct {
+	runtime *obi.Config
+}
+
+func (c *receiverConfig) Unmarshal(component *confmap.Conf) error {
+	if component == nil {
+		return nil
+	}
+
+	data, err := yaml.Marshal(component.ToStringMap())
+	if err != nil {
+		return fmt.Errorf("marshal OBI receiver config: %w", err)
+	}
+
+	extension, err := schema.ParseReceiverYAML(data)
+	if err != nil {
+		var notV2 *schema.NotV2Error
+		if !errors.As(err, &notV2) {
+			return fmt.Errorf("parse OBI receiver config v2: %w", err)
+		}
+
+		cfg := defaultRuntimeConfig()
+		if err := cfg.Unmarshal(component); err != nil {
+			return fmt.Errorf("parse legacy OBI receiver config: %w", err)
+		}
+		c.runtime = cfg
+		return nil
+	}
+
+	cfg, err := convert.V2ToRuntime(extension)
+	if err != nil {
+		return fmt.Errorf("convert OBI receiver config v2: %w", err)
+	}
+	setReceiverConsumers(cfg)
+	c.runtime = cfg
+	return nil
+}
+
+func (c *receiverConfig) Validate() error {
+	if c == nil || c.runtime == nil {
+		return errInvalidConfig
+	}
+	return c.runtime.Validate()
+}
+
+func defaultRuntimeConfig() *obi.Config {
+	cfg := obi.DefaultConfig
+	setReceiverConsumers(&cfg)
+	return &cfg
+}
+
+func setReceiverConsumers(cfg *obi.Config) {
+	cfg.Traces.TracesConsumer = consumertest.NewNop()
+	cfg.OTELMetrics.MetricsConsumer = consumertest.NewNop()
+}

collector/config_linux_test.go

@@ -0,0 +1,214 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:build linux && (amd64 || arm64)
+
+package collector
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+
+	"go.opentelemetry.io/obi/internal/config/schema"
+	"go.opentelemetry.io/obi/pkg/obi"
+)
+
+func TestReceiverConfigStructure(t *testing.T) {
+	require.NoError(t, componenttest.CheckConfigStruct(defaultConfig()))
+}
+
+func TestReceiverConfigUnmarshalV2(t *testing.T) {
+	cfg := newTestReceiverConfig(t)
+	component := confmap.NewFromStringMap(map[string]any{
+		"version": "2.0",
+		"rules": []any{
+			map[string]any{
+				"action": "include",
+				"match": map[string]any{
+					"process": map[string]any{"open_ports": "8080"},
+				},
+			},
+		},
+		"channels": map[string]any{"buffer_len": 123},
+	})
+
+	require.NoError(t, component.Unmarshal(cfg))
+	require.NoError(t, cfg.Validate())
+	require.Equal(t, 123, cfg.runtime.ChannelBufferLen)
+	require.Len(t, cfg.runtime.Discovery.Instrument, 1)
+	require.Equal(t, []int{8080}, cfg.runtime.Discovery.Instrument[0].OpenPorts.AllValues())
+	require.NotNil(t, cfg.runtime.Traces.TracesConsumer)
+	require.NotNil(t, cfg.runtime.OTELMetrics.MetricsConsumer)
+}
+
+func TestReceiverConfigUnmarshalLegacy(t *testing.T) {
+	cfg := newTestReceiverConfig(t)
+	component := confmap.NewFromStringMap(map[string]any{
+		"open_port": "8080",
+	})
+
+	require.NoError(t, component.Unmarshal(cfg))
+	require.NoError(t, cfg.Validate())
+	require.Equal(t, []int{8080}, cfg.runtime.Port.AllValues())
+}
+
+func TestReceiverConfigRejectsStandaloneSections(t *testing.T) {
+	layouts := []struct {
+		name  string
+		key   string
+		value any
+	}{
+		{name: "v2", key: "version", value: "2.0"},
+		{name: "legacy selector", key: "open_port", value: "8080"},
+	}
+	for _, layout := range layouts {
+		t.Run(layout.name, func(t *testing.T) {
+			for _, section := range []string{"enrich", "correlation", "daemon"} {
+				t.Run(section, func(t *testing.T) {
+					cfg := newTestReceiverConfig(t)
+					component := confmap.NewFromStringMap(map[string]any{
+						layout.key: layout.value,
+						section:    map[string]any{},
+					})
+
+					err := component.Unmarshal(cfg)
+
+					var notAllowed *schema.SectionNotAllowedError
+					require.ErrorAs(t, err, &notAllowed)
+					require.Equal(t, section, notAllowed.Section)
+					require.Contains(t, err.Error(), "receiver config")
+					require.Contains(t, err.Error(), "standalone mode")
+				})
+			}
+		})
+	}
+}
+
+func TestReceiverConfigDoesNotFallbackFromInvalidV2(t *testing.T) {
+	tests := []struct {
+		name      string
+		component map[string]any
+		check     func(*testing.T, error)
+	}{
+		{
+			name: "unsupported version",
+			component: map[string]any{
+				"version":            "3.0",
+				"channel_buffer_len": 123,
+			},
+			check: func(t *testing.T, err error) {
+				var unsupported *schema.UnsupportedVersionError
+				require.ErrorAs(t, err, &unsupported)
+				require.Equal(t, "3.0", unsupported.Version)
+			},
+		},
+		{
+			name: "invalid capture value",
+			component: map[string]any{
+				"version": "2.0",
+				"network": map[string]any{
+					"capture": map[string]any{"source": "invalid"},
+				},
+				"channel_buffer_len": 123,
+			},
+			check: func(t *testing.T, err error) {
+				require.Contains(t, err.Error(), "invalid source")
+				var notV2 *schema.NotV2Error
+				require.NotErrorAs(t, err, &notV2)
+			},
+		},
+		{
+			name: "standalone v2 layout with legacy selector",
+			component: map[string]any{
+				"file_format": "1.0",
+				"extensions": map[string]any{
+					"obi": map[string]any{
+						"version": "2.0",
+						"capture": map[string]any{},
+					},
+				},
+				"open_port":          "8080",
+				"channel_buffer_len": 123,
+			},
+			check: func(t *testing.T, err error) {
+				var wrongLayout *schema.ReceiverLayoutError
+				require.ErrorAs(t, err, &wrongLayout)
+				var notV2 *schema.NotV2Error
+				require.NotErrorAs(t, err, &notV2)
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := newTestReceiverConfig(t)
+
+			err := confmap.NewFromStringMap(test.component).Unmarshal(cfg)
+
+			require.Error(t, err)
+			test.check(t, err)
+			require.Equal(t, obi.DefaultConfig.ChannelBufferLen, cfg.runtime.ChannelBufferLen)
+		})
+	}
+}
+
+func TestReceiverV2PipelineModes(t *testing.T) {
+	tests := []struct {
+		name    string
+		traces  bool
+		metrics bool
+	}{
+		{name: "traces", traces: true},
+		{name: "metrics", metrics: true},
+		{name: "traces and metrics", traces: true, metrics: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := newTestReceiverConfig(t)
+			require.NoError(t, confmap.NewFromStringMap(map[string]any{
+				"version": "2.0",
+			}).Unmarshal(cfg))
+
+			factory := NewFactory()
+			settings := receivertest.NewNopSettings(typeStr)
+			settings.ID = component.MustNewIDWithName("obi", test.name)
+
+			if test.traces {
+				consumer := consumertest.NewNop()
+				receiver, err := factory.CreateTraces(t.Context(), settings, cfg, consumer)
+				require.NoError(t, err)
+				require.Same(t, consumer, cfg.runtime.Traces.TracesConsumer)
+				t.Cleanup(func() {
+					require.NoError(t, receiver.Shutdown(context.Background()))
+				})
+			}
+
+			if test.metrics {
+				consumer := consumertest.NewNop()
+				receiver, err := factory.CreateMetrics(t.Context(), settings, cfg, consumer)
+				require.NoError(t, err)
+				require.Same(t, consumer, cfg.runtime.OTELMetrics.MetricsConsumer)
+				t.Cleanup(func() {
+					require.NoError(t, receiver.Shutdown(context.Background()))
+				})
+			}
+		})
+	}
+}
+
+func newTestReceiverConfig(t *testing.T) *receiverConfig {
+	t.Helper()
+
+	cfg, ok := NewFactory().CreateDefaultConfig().(*receiverConfig)
+	require.True(t, ok)
+	return cfg
+}

collector/factory_linux.go

@@ -14,7 +14,6 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/receiver"
 
 	"go.opentelemetry.io/obi/collector/internal"
@@ -37,7 +36,7 @@ func BuildTracesReceiver() receiver.CreateTracesFunc {
 	) (receiver.Traces, error) {
 		initLogger(rs)
 
-		cfg, ok := baseCfg.(*obi.Config)
+		cfg, ok := runtimeConfig(baseCfg)
 		if !ok {
 			return nil, errInvalidConfig
 		}
@@ -55,7 +54,7 @@ func BuildMetricsReceiver() receiver.CreateMetricsFunc {
 	) (receiver.Metrics, error) {
 		initLogger(rs)
 
-		cfg, ok := baseCfg.(*obi.Config)
+		cfg, ok := runtimeConfig(baseCfg)
 		if !ok {
 			return nil, errInvalidConfig
 		}
@@ -66,10 +65,16 @@ func BuildMetricsReceiver() receiver.CreateMetricsFunc {
 }
 
 func defaultConfig() component.Config {
-	cfg := obi.DefaultConfig
-	// These are placeholders for the consumers; without these obi config will be invalid.
-	// The actual consumers are set when the receiver is created.
-	cfg.Traces.TracesConsumer = consumertest.NewNop()
-	cfg.OTELMetrics.MetricsConsumer = consumertest.NewNop()
-	return &cfg
+	return &receiverConfig{runtime: defaultRuntimeConfig()}
+}
+
+func runtimeConfig(baseCfg component.Config) (*obi.Config, bool) {
+	switch cfg := baseCfg.(type) {
+	case *receiverConfig:
+		return cfg.runtime, cfg.runtime != nil
+	case *obi.Config:
+		return cfg, cfg != nil
+	default:
+		return nil, false
+	}
 }

examples/otel-collector/README.md

@@ -41,6 +41,17 @@ The collector requires `sudo` to attach eBPF probes to processes.
 
 ## Testing the Collector
 
+After building the collector, run the same end-to-end smoke test used by the
+pull request workflow:
+
+```bash
+sudo ./smoke-test.sh
+```
+
+The smoke test starts an HTTP server and the Collector with `smoke-config.yaml`,
+then verifies that the Config v2 OBI receiver exports the server's trace through
+the debug exporter.
+
 Once the collector is running, you can generate some test traces:
 
 1. In a new terminal, start a simple HTTP server:
@@ -130,11 +141,12 @@ Once the collector is running, you can generate some test traces:
 
 The `config.yaml` file defines:
 
-- **OBI receiver**: Listens on port 8000 for HTTP traffic and automatically instruments services
+- **OBI receiver**: Uses Config v2 to capture services listening on port 8000
 - **OTLP receiver**: Accepts spans from manually instrumented applications
 - **Batch processor**: Groups spans for efficient export
 - **Debug exporter**: Prints spans to logs (useful for debugging)
 - **OTLP exporter**: Sends spans to a Jaeger backend (requires Jaeger to be running)
+- **Trace and metric pipelines**: Share one OBI receiver instance
 
 You can modify `config.yaml` to: