Move fanout consumers to fanoutconsumer package (#2615)

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+## 🛑 Breaking changes 🛑
+- Move fanout consumers to fanoutconsumer package (#2615)
+
+## 💡 Enhancements 💡
+
+## 🧰 Bug fixes 🧰
+
 ## v0.22.0 Beta
 
 ## 🛑 Breaking changes 🛑

processor/cloningfanoutconnector.go renamed to consumer/fanoutconsumer/cloningconsumer.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package processor
+package fanoutconsumer
 
 import (
 	"context"
@@ -22,27 +22,22 @@ import (
 	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
-// This file contains implementations of cloning Trace/Metrics connectors
-// that fan out the data to multiple other consumers. Cloning connectors create
-// clones of data before fanning out, which ensures each consumer gets their
-// own copy of data and is free to modify it.
-
-// NewMetricsCloningFanOutConnector wraps multiple metrics consumers in a single one and clones the data
+// NewMetricsCloning wraps multiple metrics consumers in a single one and clones the data
 // before fanning out.
-func NewMetricsCloningFanOutConnector(mcs []consumer.MetricsConsumer) consumer.MetricsConsumer {
+func NewMetricsCloning(mcs []consumer.MetricsConsumer) consumer.MetricsConsumer {
 	if len(mcs) == 1 {
 		// Don't wrap if no need to do it.
 		return mcs[0]
 	}
-	return metricsCloningFanOutConnector(mcs)
+	return metricsCloningConsumer(mcs)
 }
 
-type metricsCloningFanOutConnector []consumer.MetricsConsumer
+type metricsCloningConsumer []consumer.MetricsConsumer
 
-var _ consumer.MetricsConsumer = (*metricsCloningFanOutConnector)(nil)
+var _ consumer.MetricsConsumer = (*metricsCloningConsumer)(nil)
 
-// ConsumeMetrics exports the MetricsData to all consumers wrapped by the current one.
-func (mfc metricsCloningFanOutConnector) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
+// ConsumeMetrics exports the pdata.Metrics to all consumers wrapped by the current one.
+func (mfc metricsCloningConsumer) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
 	var errs []error
 
 	// Fan out to first len-1 consumers.
@@ -64,22 +59,22 @@ func (mfc metricsCloningFanOutConnector) ConsumeMetrics(ctx context.Context, md
 	return consumererror.CombineErrors(errs)
 }
 
-// NewTracesCloningFanOutConnector wraps multiple traces consumers in a single one and clones the data
+// NewTracesCloning wraps multiple traces consumers in a single one and clones the data
 // before fanning out.
-func NewTracesCloningFanOutConnector(tcs []consumer.TracesConsumer) consumer.TracesConsumer {
+func NewTracesCloning(tcs []consumer.TracesConsumer) consumer.TracesConsumer {
 	if len(tcs) == 1 {
 		// Don't wrap if no need to do it.
 		return tcs[0]
 	}
-	return tracesCloningFanOutConnector(tcs)
+	return tracesCloningConsumer(tcs)
 }
 
-type tracesCloningFanOutConnector []consumer.TracesConsumer
+type tracesCloningConsumer []consumer.TracesConsumer
 
-var _ consumer.TracesConsumer = (*tracesCloningFanOutConnector)(nil)
+var _ consumer.TracesConsumer = (*tracesCloningConsumer)(nil)
 
-// ConsumeTraces exports the span data to all trace consumers wrapped by the current one.
-func (tfc tracesCloningFanOutConnector) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
+// ConsumeTraces exports the pdata.Traces to all consumers wrapped by the current one.
+func (tfc tracesCloningConsumer) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
 	var errs []error
 
 	// Fan out to first len-1 consumers.
@@ -101,21 +96,22 @@ func (tfc tracesCloningFanOutConnector) ConsumeTraces(ctx context.Context, td pd
 	return consumererror.CombineErrors(errs)
 }
 
-// NewLogsCloningFanOutConnector wraps multiple trace consumers in a single one.
-func NewLogsCloningFanOutConnector(lcs []consumer.LogsConsumer) consumer.LogsConsumer {
+// NewLogsCloning wraps multiple trace consumers in a single one and clones the data
+// before fanning out.
+func NewLogsCloning(lcs []consumer.LogsConsumer) consumer.LogsConsumer {
 	if len(lcs) == 1 {
 		// Don't wrap if no need to do it.
 		return lcs[0]
 	}
-	return logsCloningFanOutConnector(lcs)
+	return logsCloningConsumer(lcs)
 }
 
-type logsCloningFanOutConnector []consumer.LogsConsumer
+type logsCloningConsumer []consumer.LogsConsumer
 
-var _ consumer.LogsConsumer = (*logsCloningFanOutConnector)(nil)
+var _ consumer.LogsConsumer = (*logsCloningConsumer)(nil)
 
-// ConsumeLogs exports the log data to all consumers wrapped by the current one.
-func (lfc logsCloningFanOutConnector) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
+// ConsumeLogs exports the pdata.Logs to all consumers wrapped by the current one.
+func (lfc logsCloningConsumer) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
 	var errs []error
 
 	// Fan out to first len-1 consumers.

processor/cloningfanoutconnector_test.go renamed to consumer/fanoutconsumer/cloningconsumer_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package processor
+package fanoutconsumer
 
 import (
 	"context"
@@ -27,7 +27,7 @@ import (
 
 func TestTraceProcessorCloningNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewTracesNop()
-	tfc := NewTracesCloningFanOutConnector([]consumer.TracesConsumer{nop})
+	tfc := NewTracesCloning([]consumer.TracesConsumer{nop})
 	assert.Same(t, nop, tfc)
 }
 
@@ -37,7 +37,7 @@ func TestTraceProcessorCloningMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.TracesSink)
 	}
 
-	tfc := NewTracesCloningFanOutConnector(processors)
+	tfc := NewTracesCloning(processors)
 	td := testdata.GenerateTraceDataTwoSpansSameResource()
 
 	var wantSpansCount = 0
@@ -70,7 +70,7 @@ func TestTraceProcessorCloningMultiplexing(t *testing.T) {
 
 func TestMetricsProcessorCloningNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewMetricsNop()
-	mfc := NewMetricsFanOutConnector([]consumer.MetricsConsumer{nop})
+	mfc := NewMetrics([]consumer.MetricsConsumer{nop})
 	assert.Same(t, nop, mfc)
 }
 
@@ -80,7 +80,7 @@ func TestMetricsProcessorCloningMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.MetricsSink)
 	}
 
-	mfc := NewMetricsCloningFanOutConnector(processors)
+	mfc := NewMetricsCloning(processors)
 	md := testdata.GeneratMetricsAllTypesWithSampleDatapoints()
 
 	var wantMetricsCount = 0
@@ -113,7 +113,7 @@ func TestMetricsProcessorCloningMultiplexing(t *testing.T) {
 
 func TestLogsProcessorCloningNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewLogsNop()
-	lfc := NewLogsCloningFanOutConnector([]consumer.LogsConsumer{nop})
+	lfc := NewLogsCloning([]consumer.LogsConsumer{nop})
 	assert.Same(t, nop, lfc)
 }
 
@@ -123,7 +123,7 @@ func TestLogsProcessorCloningMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.LogsSink)
 	}
 
-	mfc := NewLogsCloningFanOutConnector(processors)
+	mfc := NewLogsCloning(processors)
 	ld := testdata.GenerateLogDataOneLog()
 
 	var wantMetricsCount = 0

processor/fanoutconnector.go renamed to consumer/fanoutconsumer/consumer.go

@@ -12,7 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package processor
+// Package fanoutconsumer contains implementations of Traces/Metrics/Logs consumers
+// that fan out the data to multiple other consumers.
+//
+// Cloning connectors create clones of data before fanning out, which ensures each
+// consumer gets their own copy of data and is free to modify it.
+package fanoutconsumer
 
 import (
 	"context"
@@ -22,24 +27,21 @@ import (
 	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
-// This file contains implementations of Trace/Metrics connectors
-// that fan out the data to multiple other consumers.
-
-// NewMetricsFanOutConnector wraps multiple metrics consumers in a single one.
-func NewMetricsFanOutConnector(mcs []consumer.MetricsConsumer) consumer.MetricsConsumer {
+// NewMetrics wraps multiple metrics consumers in a single one.
+func NewMetrics(mcs []consumer.MetricsConsumer) consumer.MetricsConsumer {
 	if len(mcs) == 1 {
 		// Don't wrap if no need to do it.
 		return mcs[0]
 	}
-	return metricsFanOutConnector(mcs)
+	return metricsConsumer(mcs)
 }
 
-type metricsFanOutConnector []consumer.MetricsConsumer
+type metricsConsumer []consumer.MetricsConsumer
 
-var _ consumer.MetricsConsumer = (*metricsFanOutConnector)(nil)
+var _ consumer.MetricsConsumer = (*metricsConsumer)(nil)
 
 // ConsumeMetrics exports the pdata.Metrics to all consumers wrapped by the current one.
-func (mfc metricsFanOutConnector) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
+func (mfc metricsConsumer) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
 	var errs []error
 	for _, mc := range mfc {
 		if err := mc.ConsumeMetrics(ctx, md); err != nil {
@@ -49,21 +51,21 @@ func (mfc metricsFanOutConnector) ConsumeMetrics(ctx context.Context, md pdata.M
 	return consumererror.CombineErrors(errs)
 }
 
-// NewTracesFanOutConnector wraps multiple trace consumers in a single one.
-func NewTracesFanOutConnector(tcs []consumer.TracesConsumer) consumer.TracesConsumer {
+// NewTraces wraps multiple trace consumers in a single one.
+func NewTraces(tcs []consumer.TracesConsumer) consumer.TracesConsumer {
 	if len(tcs) == 1 {
 		// Don't wrap if no need to do it.
 		return tcs[0]
 	}
-	return traceFanOutConnector(tcs)
+	return traceConsumer(tcs)
 }
 
-type traceFanOutConnector []consumer.TracesConsumer
+type traceConsumer []consumer.TracesConsumer
 
-var _ consumer.TracesConsumer = (*traceFanOutConnector)(nil)
+var _ consumer.TracesConsumer = (*traceConsumer)(nil)
 
-// ConsumeTraces exports the span data to all trace consumers wrapped by the current one.
-func (tfc traceFanOutConnector) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
+// ConsumeTraces exports the pdata.Traces to all consumers wrapped by the current one.
+func (tfc traceConsumer) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
 	var errs []error
 	for _, tc := range tfc {
 		if err := tc.ConsumeTraces(ctx, td); err != nil {
@@ -73,21 +75,21 @@ func (tfc traceFanOutConnector) ConsumeTraces(ctx context.Context, td pdata.Trac
 	return consumererror.CombineErrors(errs)
 }
 
-// NewLogsFanOutConnector wraps multiple log consumers in a single one.
-func NewLogsFanOutConnector(lcs []consumer.LogsConsumer) consumer.LogsConsumer {
+// NewLogs wraps multiple log consumers in a single one.
+func NewLogs(lcs []consumer.LogsConsumer) consumer.LogsConsumer {
 	if len(lcs) == 1 {
 		// Don't wrap if no need to do it.
 		return lcs[0]
 	}
-	return logsFanOutConnector(lcs)
+	return logsConsumer(lcs)
 }
 
-type logsFanOutConnector []consumer.LogsConsumer
+type logsConsumer []consumer.LogsConsumer
 
-var _ consumer.LogsConsumer = (*logsFanOutConnector)(nil)
+var _ consumer.LogsConsumer = (*logsConsumer)(nil)
 
-// Consume exports the log data to all consumers wrapped by the current one.
-func (fc logsFanOutConnector) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
+// ConsumeLogs exports the pdata.Logs to all consumers wrapped by the current one.
+func (fc logsConsumer) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
 	var errs []error
 	for _, tc := range fc {
 		if err := tc.ConsumeLogs(ctx, ld); err != nil {

processor/fanoutconnector_test.go renamed to consumer/fanoutconsumer/consumer_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package processor
+package fanoutconsumer
 
 import (
 	"context"
@@ -28,7 +28,7 @@ import (
 
 func TestTracesProcessorNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewTracesNop()
-	tfc := NewTracesFanOutConnector([]consumer.TracesConsumer{nop})
+	tfc := NewTraces([]consumer.TracesConsumer{nop})
 	assert.Same(t, nop, tfc)
 }
 
@@ -38,7 +38,7 @@ func TestTracesProcessorMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.TracesSink)
 	}
 
-	tfc := NewTracesFanOutConnector(processors)
+	tfc := NewTraces(processors)
 	td := testdata.GenerateTraceDataOneSpan()
 
 	var wantSpansCount = 0
@@ -67,7 +67,7 @@ func TestTraceProcessorWhenOneErrors(t *testing.T) {
 	// Make one processor return error
 	processors[1] = consumertest.NewTracesErr(errors.New("my error"))
 
-	tfc := NewTracesFanOutConnector(processors)
+	tfc := NewTraces(processors)
 	td := testdata.GenerateTraceDataOneSpan()
 
 	var wantSpansCount = 0
@@ -82,7 +82,7 @@ func TestTraceProcessorWhenOneErrors(t *testing.T) {
 
 func TestMetricsProcessorNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewMetricsNop()
-	mfc := NewMetricsFanOutConnector([]consumer.MetricsConsumer{nop})
+	mfc := NewMetrics([]consumer.MetricsConsumer{nop})
 	assert.Same(t, nop, mfc)
 }
 
@@ -92,7 +92,7 @@ func TestMetricsProcessorMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.MetricsSink)
 	}
 
-	mfc := NewMetricsFanOutConnector(processors)
+	mfc := NewMetrics(processors)
 	md := testdata.GenerateMetricsOneMetric()
 
 	var wantMetricsCount = 0
@@ -121,7 +121,7 @@ func TestMetricsProcessorWhenOneErrors(t *testing.T) {
 	// Make one processor return error
 	processors[1] = consumertest.NewMetricsErr(errors.New("my error"))
 
-	mfc := NewMetricsFanOutConnector(processors)
+	mfc := NewMetrics(processors)
 	md := testdata.GenerateMetricsOneMetric()
 
 	var wantMetricsCount = 0
@@ -136,7 +136,7 @@ func TestMetricsProcessorWhenOneErrors(t *testing.T) {
 
 func TestLogsProcessorNotMultiplexing(t *testing.T) {
 	nop := consumertest.NewLogsNop()
-	lfc := NewLogsFanOutConnector([]consumer.LogsConsumer{nop})
+	lfc := NewLogs([]consumer.LogsConsumer{nop})
 	assert.Same(t, nop, lfc)
 }
 
@@ -146,7 +146,7 @@ func TestLogsProcessorMultiplexing(t *testing.T) {
 		processors[i] = new(consumertest.LogsSink)
 	}
 
-	lfc := NewLogsFanOutConnector(processors)
+	lfc := NewLogs(processors)
 	ld := testdata.GenerateLogDataOneLog()
 
 	var wantMetricsCount = 0
@@ -175,7 +175,7 @@ func TestLogsProcessorWhenOneErrors(t *testing.T) {
 	// Make one processor return error
 	processors[1] = consumertest.NewLogsErr(errors.New("my error"))
 
-	lfc := NewLogsFanOutConnector(processors)
+	lfc := NewLogs(processors)
 	ld := testdata.GenerateLogDataOneLog()
 
 	var wantMetricsCount = 0