Raw import @ c86a6c7

receiver/otelarrowreceiver/config.go

@@ -9,12 +9,19 @@ import (
 	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+const (
+	// Confmap values.
+	protoGRPC                = "protocols::grpc"
+	protoArrowMemoryLimitMiB = "protocols::arrow::memory_limit_mib"
 )
 
 // Protocols is the configuration for the supported protocols.
 type Protocols struct {
 	GRPC  configgrpc.ServerConfig `mapstructure:"grpc"`
-	Arrow ArrowSettings           `mapstructure:"arrow"`
+	Arrow ArrowSettings                 `mapstructure:"arrow"`
 }
 
 // ArrowSettings support configuring the Arrow receiver.
@@ -35,6 +42,7 @@ type Config struct {
 }
 
 var _ component.Config = (*Config)(nil)
+var _ confmap.Unmarshaler = (*Config)(nil)
 
 // Validate checks the receiver configuration is valid
 func (cfg *Config) Validate() error {
@@ -50,3 +58,14 @@ func (cfg *ArrowSettings) Validate() error {
 	}
 	return nil
 }
+
+// Unmarshal a confmap.Conf into the config struct.
+func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
+	// first load the config normally
+	err := conf.Unmarshal(cfg)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

receiver/otelarrowreceiver/config_test.go

@@ -8,9 +8,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/confignet"
@@ -51,10 +51,10 @@ func TestUnmarshalConfig(t *testing.T) {
 				GRPC: configgrpc.ServerConfig{
 					NetAddr: confignet.AddrConfig{
 						Endpoint:  "0.0.0.0:4317",
-						Transport: confignet.TransportTypeTCP,
+						Transport: "tcp",
 					},
-					TLSSetting: &configtls.ServerConfig{
-						TLSSetting: configtls.Config{
+					TLSSetting: &configtls.TLSServerSetting{
+						TLSSetting: configtls.TLSSetting{
 							CertFile: "test.crt",
 							KeyFile:  "test.key",
 						},
@@ -79,9 +79,6 @@ func TestUnmarshalConfig(t *testing.T) {
 				},
 				Arrow: ArrowSettings{
 					MemoryLimitMiB: 123,
-					Zstd: zstd.DecoderConfig{
-						MemoryLimitMiB: 8,
-					},
 				},
 			},
 		}, cfg)
@@ -100,7 +97,7 @@ func TestUnmarshalConfigUnix(t *testing.T) {
 				GRPC: configgrpc.ServerConfig{
 					NetAddr: confignet.AddrConfig{
 						Endpoint:  "/tmp/grpc_otlp.sock",
-						Transport: confignet.TransportTypeUnix,
+						Transport: "unix",
 					},
 					ReadBufferSize: 512 * 1024,
 				},
@@ -111,6 +108,14 @@ func TestUnmarshalConfigUnix(t *testing.T) {
 		}, cfg)
 }
 
+func TestUnmarshalConfigTypoDefaultProtocol(t *testing.T) {
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "typo_default_proto_config.yaml"))
+	require.NoError(t, err)
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.EqualError(t, component.UnmarshalConfig(cm, cfg), "1 error(s) decoding:\n\n* 'protocols' has invalid keys: htttp")
+}
+
 func TestUnmarshalConfigInvalidProtocol(t *testing.T) {
 	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "bad_proto_config.yaml"))
 	require.NoError(t, err)
@@ -121,5 +126,7 @@ func TestUnmarshalConfigInvalidProtocol(t *testing.T) {
 
 func TestUnmarshalConfigNoProtocols(t *testing.T) {
 	cfg := Config{}
-	assert.Error(t, component.ValidateConfig(cfg))
+	// This now produces an error due to breaking change.
+	// https://github.com/open-telemetry/opentelemetry-collector/pull/9385
+	assert.ErrorContains(t, component.ValidateConfig(cfg), "invalid transport type")
 }

receiver/otelarrowreceiver/factory.go

@@ -6,14 +6,13 @@ package otelarrowreceiver // import "github.com/open-telemetry/opentelemetry-col
 import (
 	"context"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/metadata"
+	"github.com/open-telemetry/otel-arrow/collector/sharedcomponent"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/metadata"
 )
 
 const (
@@ -22,10 +21,10 @@ const (
 	defaultMemoryLimitMiB = 128
 )
 
-// NewFactory creates a new OTel-Arrow receiver factory.
+// NewFactory creates a new OTLP receiver factory.
 func NewFactory() receiver.Factory {
 	return receiver.NewFactory(
-		metadata.Type,
+		component.MustNewType(metadata.Type),
 		createDefaultConfig,
 		receiver.WithTraces(createTraces, metadata.TracesStability),
 		receiver.WithMetrics(createMetrics, metadata.MetricsStability),
@@ -39,7 +38,7 @@ func createDefaultConfig() component.Config {
 			GRPC: configgrpc.ServerConfig{
 				NetAddr: confignet.AddrConfig{
 					Endpoint:  defaultGRPCEndpoint,
-					Transport: confignet.TransportTypeTCP,
+					Transport: "tcp",
 				},
 				// We almost write 0 bytes, so no need to tune WriteBufferSize.
 				ReadBufferSize: 512 * 1024,
@@ -57,17 +56,18 @@ func createTraces(
 	set receiver.CreateSettings,
 	cfg component.Config,
 	nextConsumer consumer.Traces,
-) (_ receiver.Traces, err error) {
+) (receiver.Traces, error) {
 	oCfg := cfg.(*Config)
-	r := receivers.GetOrAdd(oCfg, func() (comp component.Component) {
-		comp, err = newOTelArrowReceiver(oCfg, set)
-		return
+	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
+		return newOTelArrowReceiver(oCfg, set)
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	r.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer)
+	if err = r.Unwrap().registerTraceConsumer(nextConsumer); err != nil {
+		return nil, err
+	}
 	return r, nil
 }
 
@@ -77,17 +77,18 @@ func createMetrics(
 	set receiver.CreateSettings,
 	cfg component.Config,
 	consumer consumer.Metrics,
-) (_ receiver.Metrics, err error) {
+) (receiver.Metrics, error) {
 	oCfg := cfg.(*Config)
-	r := receivers.GetOrAdd(oCfg, func() (comp component.Component) {
-		comp, err = newOTelArrowReceiver(oCfg, set)
-		return comp
+	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
+		return newOTelArrowReceiver(oCfg, set)
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	r.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer)
+	if err = r.Unwrap().registerMetricsConsumer(consumer); err != nil {
+		return nil, err
+	}
 	return r, nil
 }
 
@@ -97,17 +98,18 @@ func createLog(
 	set receiver.CreateSettings,
 	cfg component.Config,
 	consumer consumer.Logs,
-) (_ receiver.Logs, err error) {
+) (receiver.Logs, error) {
 	oCfg := cfg.(*Config)
-	r := receivers.GetOrAdd(oCfg, func() (comp component.Component) {
-		comp, err = newOTelArrowReceiver(oCfg, set)
-		return comp
+	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
+		return newOTelArrowReceiver(oCfg, set)
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	r.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer)
+	if err = r.Unwrap().registerLogsConsumer(consumer); err != nil {
+		return nil, err
+	}
 	return r, nil
 }
 
@@ -117,4 +119,4 @@ func createLog(
 // create separate objects, they must use one otelArrowReceiver object per configuration.
 // When the receiver is shutdown it should be removed from this map so the same configuration
 // can be recreated successfully.
-var receivers = sharedcomponent.NewSharedComponents()
+var receivers = sharedcomponent.NewSharedComponents[*Config, *otelArrowReceiver]()

receiver/otelarrowreceiver/factory_test.go

@@ -7,9 +7,10 @@ import (
 	"context"
 	"testing"
 
-	"github.com/open-telemetry/otel-arrow/collector/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/open-telemetry/otel-arrow/collector/testutil"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/confignet"
@@ -45,7 +46,7 @@ func TestCreateTracesReceiver(t *testing.T) {
 	defaultGRPCSettings := configgrpc.ServerConfig{
 		NetAddr: confignet.AddrConfig{
 			Endpoint:  testutil.GetAvailableLocalAddress(t),
-			Transport: confignet.TransportTypeTCP,
+			Transport: "tcp",
 		},
 	}
 
@@ -69,7 +70,7 @@ func TestCreateTracesReceiver(t *testing.T) {
 					GRPC: configgrpc.ServerConfig{
 						NetAddr: confignet.AddrConfig{
 							Endpoint:  "localhost:112233",
-							Transport: confignet.TransportTypeTCP,
+							Transport: "tcp",
 						},
 					},
 				},
@@ -101,7 +102,7 @@ func TestCreateMetricReceiver(t *testing.T) {
 	defaultGRPCSettings := configgrpc.ServerConfig{
 		NetAddr: confignet.AddrConfig{
 			Endpoint:  testutil.GetAvailableLocalAddress(t),
-			Transport: confignet.TransportTypeTCP,
+			Transport: "tcp",
 		},
 	}
 
@@ -125,7 +126,7 @@ func TestCreateMetricReceiver(t *testing.T) {
 					GRPC: configgrpc.ServerConfig{
 						NetAddr: confignet.AddrConfig{
 							Endpoint:  "327.0.0.1:1122",
-							Transport: confignet.TransportTypeTCP,
+							Transport: "tcp",
 						},
 					},
 				},
@@ -156,7 +157,7 @@ func TestCreateLogReceiver(t *testing.T) {
 	defaultGRPCSettings := configgrpc.ServerConfig{
 		NetAddr: confignet.AddrConfig{
 			Endpoint:  testutil.GetAvailableLocalAddress(t),
-			Transport: confignet.TransportTypeTCP,
+			Transport: "tcp",
 		},
 	}
 
@@ -183,7 +184,7 @@ func TestCreateLogReceiver(t *testing.T) {
 					GRPC: configgrpc.ServerConfig{
 						NetAddr: confignet.AddrConfig{
 							Endpoint:  "327.0.0.1:1122",
-							Transport: confignet.TransportTypeTCP,
+							Transport: "tcp",
 						},
 					},
 				},