otelarrow receiver

receiver/otelarrowreceiver/config.go

@@ -6,7 +6,7 @@ package otelarrowreceiver // import "github.com/open-telemetry/opentelemetry-col
 import (
 	"fmt"
 
-	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/compression/zstd"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
 )

receiver/otelarrowreceiver/factory.go

@@ -6,7 +6,7 @@ package otelarrowreceiver // import "github.com/open-telemetry/opentelemetry-col
 import (
 	"context"
 
-	"github.com/open-telemetry/otel-arrow/collector/sharedcomponent"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/confignet"
@@ -24,18 +24,25 @@ const (
 	defaultWaiterLimit       = 1000
 )
 
+type receiverFactory struct {
+	receivers *sharedcomponent.SharedComponents
+}
+
 // NewFactory creates a new OTel-Arrow receiver factory.
 func NewFactory() receiver.Factory {
+	f := &receiverFactory{
+		receivers: sharedcomponent.NewSharedComponents(),
+	}
 	return receiver.NewFactory(
 		metadata.Type,
-		createDefaultConfig,
-		receiver.WithTraces(createTraces, metadata.TracesStability),
-		receiver.WithMetrics(createMetrics, metadata.MetricsStability),
-		receiver.WithLogs(createLog, metadata.LogsStability))
+		f.createDefaultConfig,
+		receiver.WithTraces(f.createTraces, metadata.TracesStability),
+		receiver.WithMetrics(f.createMetrics, metadata.MetricsStability),
+		receiver.WithLogs(f.createLog, metadata.LogsStability))
 }
 
 // createDefaultConfig creates the default configuration for receiver.
-func createDefaultConfig() component.Config {
+func (f *receiverFactory) createDefaultConfig() component.Config {
 	return &Config{
 		Protocols: Protocols{
 			GRPC: configgrpc.ServerConfig{
@@ -56,59 +63,68 @@ func createDefaultConfig() component.Config {
 }
 
 // createTraces creates a trace receiver based on provided config.
-func createTraces(
+func (f *receiverFactory) createTraces(
 	_ context.Context,
 	set receiver.Settings,
 	cfg component.Config,
 	nextConsumer consumer.Traces,
 ) (receiver.Traces, error) {
 	oCfg := cfg.(*Config)
-	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
-		return newOTelArrowReceiver(oCfg, set)
+	var compErr error
+	r := f.receivers.GetOrAdd(oCfg, func() component.Component {
+		var comp component.Component
+		comp, compErr = newOTelArrowReceiver(oCfg, set)
+		return comp
 	})
-	if err != nil {
-		return nil, err
+	if compErr != nil {
+		return nil, compErr
 	}
 
-	r.Unwrap().registerTraceConsumer(nextConsumer)
+	r.Unwrap().(*otelArrowReceiver).registerTraceConsumer(nextConsumer)
 	return r, nil
 }
 
 // createMetrics creates a metrics receiver based on provided config.
-func createMetrics(
+func (f *receiverFactory) createMetrics(
 	_ context.Context,
 	set receiver.Settings,
 	cfg component.Config,
 	consumer consumer.Metrics,
 ) (receiver.Metrics, error) {
 	oCfg := cfg.(*Config)
-	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
-		return newOTelArrowReceiver(oCfg, set)
+	var compErr error
+	r := f.receivers.GetOrAdd(oCfg, func() component.Component {
+		var comp component.Component
+		comp, compErr = newOTelArrowReceiver(oCfg, set)
+		return comp
 	})
-	if err != nil {
-		return nil, err
+	if compErr != nil {
+		return nil, compErr
 	}
 
-	r.Unwrap().registerMetricsConsumer(consumer)
+	r.Unwrap().(*otelArrowReceiver).registerMetricsConsumer(consumer)
 	return r, nil
 }
 
 // createLog creates a log receiver based on provided config.
-func createLog(
+func (f *receiverFactory) createLog(
 	_ context.Context,
 	set receiver.Settings,
 	cfg component.Config,
 	consumer consumer.Logs,
 ) (receiver.Logs, error) {
 	oCfg := cfg.(*Config)
-	r, err := receivers.GetOrAdd(oCfg, func() (*otelArrowReceiver, error) {
-		return newOTelArrowReceiver(oCfg, set)
+	var compErr error
+	r := f.receivers.GetOrAdd(oCfg, func() component.Component {
+		var comp component.Component
+		comp, compErr = newOTelArrowReceiver(oCfg, set)
+		return comp
 	})
-	if err != nil {
-		return nil, err
+	if compErr != nil {
+		return nil, compErr
 	}
 
-	r.Unwrap().registerLogsConsumer(consumer)
+	r.Unwrap().(*otelArrowReceiver).registerLogsConsumer(consumer)
 	return r, nil
 }
 
@@ -118,4 +134,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[*Config, *otelArrowReceiver]()
+var receivers = sharedcomponent.NewSharedComponents()

receiver/otelarrowreceiver/factory_test.go

@@ -7,7 +7,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/open-telemetry/otel-arrow/collector/testutil"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"

receiver/otelarrowreceiver/go.mod

@@ -3,8 +3,9 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelar
 go 1.21.0
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow v0.102.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.102.0
 	github.com/open-telemetry/otel-arrow v0.24.0
-	github.com/open-telemetry/otel-arrow/collector v0.24.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector v0.102.2-0.20240611143128-7dfb57b9ad1c
 	go.opentelemetry.io/collector/component v0.102.2-0.20240611143128-7dfb57b9ad1c
@@ -17,7 +18,7 @@ require (
 	go.opentelemetry.io/collector/consumer v0.102.2-0.20240611143128-7dfb57b9ad1c
 	go.opentelemetry.io/collector/extension/auth v0.102.2-0.20240611143128-7dfb57b9ad1c
 	go.opentelemetry.io/collector/pdata v1.9.1-0.20240611143128-7dfb57b9ad1c
-	go.opentelemetry.io/collector/receiver v0.102.2-0.20240611143128-7dfb57b9ad1c
+	go.opentelemetry.io/collector/receiver v0.102.2-0.20240614160340-42b61cc60405
 	go.opentelemetry.io/otel v1.27.0
 	go.opentelemetry.io/otel/metric v1.27.0
 	go.opentelemetry.io/otel/sdk/metric v1.27.0
@@ -53,7 +54,7 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/go-version v1.7.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/compress v1.17.8 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
@@ -76,7 +77,7 @@ require (
 	go.opentelemetry.io/collector/config/configcompression v1.9.1-0.20240611143128-7dfb57b9ad1c // indirect
 	go.opentelemetry.io/collector/config/configopaque v1.9.1-0.20240611143128-7dfb57b9ad1c // indirect
 	go.opentelemetry.io/collector/config/internal v0.102.2-0.20240611143128-7dfb57b9ad1c // indirect
-	go.opentelemetry.io/collector/exporter v0.102.2-0.20240611143128-7dfb57b9ad1c // indirect
+	go.opentelemetry.io/collector/exporter v0.102.2-0.20240614160340-42b61cc60405 // indirect
 	go.opentelemetry.io/collector/extension v0.102.2-0.20240611143128-7dfb57b9ad1c // indirect
 	go.opentelemetry.io/collector/featuregate v1.9.1-0.20240611143128-7dfb57b9ad1c // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.52.0 // indirect
@@ -92,3 +93,5 @@ require (
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240520151616-dc85e6b867a5 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow => ../../internal/otelarrow

receiver/otelarrowreceiver/internal/arrow/arrow.go

@@ -15,9 +15,9 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/netstats"
 	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
-	"github.com/open-telemetry/otel-arrow/collector/admission"
-	"github.com/open-telemetry/otel-arrow/collector/netstats"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"

receiver/otelarrowreceiver/internal/arrow/arrow_test.go

@@ -15,11 +15,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/netstats"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
 	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
 	arrowCollectorMock "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1/mock"
-	"github.com/open-telemetry/otel-arrow/collector/admission"
-	"github.com/open-telemetry/otel-arrow/collector/netstats"
-	"github.com/open-telemetry/otel-arrow/collector/testdata"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	arrowRecordMock "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record/mock"
 	otelAssert "github.com/open-telemetry/otel-arrow/pkg/otel/assert"

receiver/otelarrowreceiver/internal/logs/otlp_test.go

@@ -9,7 +9,7 @@ import (
 	"net"
 	"testing"
 
-	"github.com/open-telemetry/otel-arrow/collector/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"

receiver/otelarrowreceiver/internal/metrics/otlp_test.go

@@ -9,7 +9,7 @@ import (
 	"net"
 	"testing"
 
-	"github.com/open-telemetry/otel-arrow/collector/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"

receiver/otelarrowreceiver/internal/trace/otlp_test.go

@@ -9,7 +9,7 @@ import (
 	"net"
 	"testing"
 
-	"github.com/open-telemetry/otel-arrow/collector/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"

receiver/otelarrowreceiver/otelarrow.go

@@ -8,10 +8,10 @@ import (
 	"errors"
 	"sync"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/compression/zstd"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/netstats"
 	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
-	"github.com/open-telemetry/otel-arrow/collector/admission"
-	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
-	"github.com/open-telemetry/otel-arrow/collector/netstats"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configgrpc"

receiver/otelarrowreceiver/otelarrow_test.go

@@ -14,9 +14,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testutil"
 	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
-	"github.com/open-telemetry/otel-arrow/collector/testdata"
-	"github.com/open-telemetry/otel-arrow/collector/testutil"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"