Deprecate SinkExporter, add SinkConsumer

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

consumer/consumertest/sink.go

@@ -0,0 +1,183 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package consumertest
+
+import (
+	"context"
+	"sync"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+type baseErrorConsumer struct {
+	mu           sync.Mutex
+	consumeError error // to be returned by ConsumeTraces, if set
+}
+
+// SetConsumeError sets an error that will be returned by the Consume function.
+func (bec *baseErrorConsumer) SetConsumeError(err error) {
+	bec.mu.Lock()
+	defer bec.mu.Unlock()
+	bec.consumeError = err
+}
+
+// SinkTraces acts as a trace receiver for use in tests.
+type SinkTraces struct {
+	baseErrorConsumer
+	traces     []pdata.Traces
+	spansCount int
+}
+
+var _ consumer.TraceConsumer = (*SinkTraces)(nil)
+
+// ConsumeTraceData stores traces for tests.
+func (ste *SinkTraces) ConsumeTraces(_ context.Context, td pdata.Traces) error {
+	ste.mu.Lock()
+	defer ste.mu.Unlock()
+
+	if ste.consumeError != nil {
+		return ste.consumeError
+	}
+
+	ste.traces = append(ste.traces, td)
+	ste.spansCount += td.SpanCount()
+
+	return nil
+}
+
+// AllTraces returns the traces sent to the test sink.
+func (ste *SinkTraces) AllTraces() []pdata.Traces {
+	ste.mu.Lock()
+	defer ste.mu.Unlock()
+
+	copyTraces := make([]pdata.Traces, len(ste.traces))
+	copy(copyTraces, ste.traces)
+	return copyTraces
+}
+
+// SpansCount return the number of spans sent to the test sing.
+func (ste *SinkTraces) SpansCount() int {
+	ste.mu.Lock()
+	defer ste.mu.Unlock()
+	return ste.spansCount
+}
+
+// Reset deletes any existing metrics.
+func (ste *SinkTraces) Reset() {
+	ste.mu.Lock()
+	defer ste.mu.Unlock()
+
+	ste.traces = nil
+	ste.spansCount = 0
+}
+
+// SinkMetrics acts as a metrics receiver for use in tests.
+type SinkMetrics struct {
+	baseErrorConsumer
+	metrics      []pdata.Metrics
+	metricsCount int
+}
+
+var _ consumer.MetricsConsumer = (*SinkMetrics)(nil)
+
+// ConsumeMetricsData stores traces for tests.
+func (sme *SinkMetrics) ConsumeMetrics(_ context.Context, md pdata.Metrics) error {
+	sme.mu.Lock()
+	defer sme.mu.Unlock()
+	if sme.consumeError != nil {
+		return sme.consumeError
+	}
+
+	sme.metrics = append(sme.metrics, md)
+	sme.metricsCount += md.MetricCount()
+
+	return nil
+}
+
+// AllMetrics returns the metrics sent to the test sink.
+func (sme *SinkMetrics) AllMetrics() []pdata.Metrics {
+	sme.mu.Lock()
+	defer sme.mu.Unlock()
+
+	copyMetrics := make([]pdata.Metrics, len(sme.metrics))
+	copy(copyMetrics, sme.metrics)
+	return copyMetrics
+}
+
+// MetricsCount return the number of metrics sent to the test sing.
+func (sme *SinkMetrics) MetricsCount() int {
+	sme.mu.Lock()
+	defer sme.mu.Unlock()
+	return sme.metricsCount
+}
+
+// Reset deletes any existing metrics.
+func (sme *SinkMetrics) Reset() {
+	sme.mu.Lock()
+	defer sme.mu.Unlock()
+
+	sme.metrics = nil
+	sme.metricsCount = 0
+}
+
+// SinkLogs acts as a metrics receiver for use in tests.
+type SinkLogs struct {
+	baseErrorConsumer
+	logs            []pdata.Logs
+	logRecordsCount int
+}
+
+var _ consumer.LogsConsumer = (*SinkLogs)(nil)
+
+// ConsumeLogData stores traces for tests.
+func (sle *SinkLogs) ConsumeLogs(_ context.Context, ld pdata.Logs) error {
+	sle.mu.Lock()
+	defer sle.mu.Unlock()
+	if sle.consumeError != nil {
+		return sle.consumeError
+	}
+
+	sle.logs = append(sle.logs, ld)
+	sle.logRecordsCount += ld.LogRecordCount()
+
+	return nil
+}
+
+// AllLog returns the metrics sent to the test sink.
+func (sle *SinkLogs) AllLogs() []pdata.Logs {
+	sle.mu.Lock()
+	defer sle.mu.Unlock()
+
+	copyLogs := make([]pdata.Logs, len(sle.logs))
+	copy(copyLogs, sle.logs)
+	return copyLogs
+}
+
+// LogRecordsCount return the number of log records sent to the test sing.
+func (sle *SinkLogs) LogRecordsCount() int {
+	sle.mu.Lock()
+	defer sle.mu.Unlock()
+	return sle.logRecordsCount
+}
+
+// Reset deletes any existing logs.
+func (sle *SinkLogs) Reset() {
+	sle.mu.Lock()
+	defer sle.mu.Unlock()
+
+	sle.logs = nil
+	sle.logRecordsCount = 0
+}

consumer/consumertest/sink_test.go

@@ -0,0 +1,99 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package consumertest
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal/data/testdata"
+)
+
+func TestSinkTraces(t *testing.T) {
+	sink := new(SinkTraces)
+	td := testdata.GenerateTraceDataOneSpan()
+	want := make([]pdata.Traces, 0, 7)
+	for i := 0; i < 7; i++ {
+		require.NoError(t, sink.ConsumeTraces(context.Background(), td))
+		want = append(want, td)
+	}
+	assert.Equal(t, want, sink.AllTraces())
+	assert.Equal(t, len(want), sink.SpansCount())
+	sink.Reset()
+	assert.Equal(t, 0, len(sink.AllTraces()))
+	assert.Equal(t, 0, sink.SpansCount())
+}
+
+func TestSinkTraces_Error(t *testing.T) {
+	sink := new(SinkTraces)
+	sink.SetConsumeError(errors.New("my error"))
+	td := testdata.GenerateTraceDataOneSpan()
+	require.Error(t, sink.ConsumeTraces(context.Background(), td))
+	assert.Len(t, sink.AllTraces(), 0)
+	assert.Equal(t, 0, sink.SpansCount())
+}
+
+func TestSinkMetrics(t *testing.T) {
+	sink := new(SinkMetrics)
+	md := testdata.GenerateMetricsOneMetric()
+	want := make([]pdata.Metrics, 0, 7)
+	for i := 0; i < 7; i++ {
+		require.NoError(t, sink.ConsumeMetrics(context.Background(), md))
+		want = append(want, md)
+	}
+	assert.Equal(t, want, sink.AllMetrics())
+	assert.Equal(t, len(want), sink.MetricsCount())
+	sink.Reset()
+	assert.Equal(t, 0, len(sink.AllMetrics()))
+	assert.Equal(t, 0, sink.MetricsCount())
+}
+
+func TestSinkMetrics_Error(t *testing.T) {
+	sink := new(SinkMetrics)
+	sink.SetConsumeError(errors.New("my error"))
+	md := testdata.GenerateMetricsOneMetric()
+	require.Error(t, sink.ConsumeMetrics(context.Background(), md))
+	assert.Len(t, sink.AllMetrics(), 0)
+	assert.Equal(t, 0, sink.MetricsCount())
+}
+
+func TestSinkLogs(t *testing.T) {
+	sink := new(SinkLogs)
+	md := testdata.GenerateLogDataOneLogNoResource()
+	want := make([]pdata.Logs, 0, 7)
+	for i := 0; i < 7; i++ {
+		require.NoError(t, sink.ConsumeLogs(context.Background(), md))
+		want = append(want, md)
+	}
+	assert.Equal(t, want, sink.AllLogs())
+	assert.Equal(t, len(want), sink.LogRecordsCount())
+	sink.Reset()
+	assert.Equal(t, 0, len(sink.AllLogs()))
+	assert.Equal(t, 0, sink.LogRecordsCount())
+}
+
+func TestSinkLogs_Error(t *testing.T) {
+	sink := new(SinkLogs)
+	sink.SetConsumeError(errors.New("my error"))
+	ld := testdata.GenerateLogDataOneLogNoResource()
+	require.Error(t, sink.ConsumeLogs(context.Background(), ld))
+	assert.Len(t, sink.AllLogs(), 0)
+	assert.Equal(t, 0, sink.LogRecordsCount())
+}

exporter/opencensusexporter/opencensus_test.go

@@ -26,15 +26,15 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/configtls"
+	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/consumer/pdata"
-	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/internal/data/testdata"
 	"go.opentelemetry.io/collector/receiver/opencensusreceiver"
 	"go.opentelemetry.io/collector/testutil"
 )
 
 func TestSendTraces(t *testing.T) {
-	sink := &exportertest.SinkTraceExporter{}
+	sink := new(consumertest.SinkTraces)
 	rFactory := opencensusreceiver.NewFactory()
 	rCfg := rFactory.CreateDefaultConfig().(*opencensusreceiver.Config)
 	endpoint := testutil.GetAvailableLocalAddress(t)
@@ -133,7 +133,7 @@ func TestSendTraces_AfterStop(t *testing.T) {
 }
 
 func TestSendMetrics(t *testing.T) {
-	sink := &exportertest.SinkMetricsExporter{}
+	sink := new(consumertest.SinkMetrics)
 	rFactory := opencensusreceiver.NewFactory()
 	rCfg := rFactory.CreateDefaultConfig().(*opencensusreceiver.Config)
 	endpoint := testutil.GetAvailableLocalAddress(t)

exporter/otlphttpexporter/otlp_test.go

@@ -30,7 +30,7 @@ import (
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/exporter/exportertest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/internal/data/testdata"
 	"go.opentelemetry.io/collector/receiver/otlpreceiver"
 	"go.opentelemetry.io/collector/testutil"
@@ -75,8 +75,8 @@ func TestTraceInvalidUrl(t *testing.T) {
 func TestTraceError(t *testing.T) {
 	addr := testutil.GetAvailableLocalAddress(t)
 
-	sink := new(exportertest.SinkTraceExporter)
-	sink.SetConsumeTraceError(errors.New("my_error"))
+	sink := new(consumertest.SinkTraces)
+	sink.SetConsumeError(errors.New("my_error"))
 	startTraceReceiver(t, addr, sink)
 	exp := startTraceExporter(t, "", fmt.Sprintf("http://%s/v1/trace", addr))
 
@@ -113,7 +113,7 @@ func TestTraceRoundTrip(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			sink := new(exportertest.SinkTraceExporter)
+			sink := new(consumertest.SinkTraces)
 			startTraceReceiver(t, addr, sink)
 			exp := startTraceExporter(t, test.baseURL, test.overrideURL)
 
@@ -132,8 +132,8 @@ func TestTraceRoundTrip(t *testing.T) {
 func TestMetricsError(t *testing.T) {
 	addr := testutil.GetAvailableLocalAddress(t)
 
-	sink := new(exportertest.SinkMetricsExporter)
-	sink.SetConsumeMetricsError(errors.New("my_error"))
+	sink := new(consumertest.SinkMetrics)
+	sink.SetConsumeError(errors.New("my_error"))
 	startMetricsReceiver(t, addr, sink)
 	exp := startMetricsExporter(t, "", fmt.Sprintf("http://%s/v1/metrics", addr))
 
@@ -168,7 +168,7 @@ func TestMetricsRoundTrip(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			sink := new(exportertest.SinkMetricsExporter)
+			sink := new(consumertest.SinkMetrics)
 			startMetricsReceiver(t, addr, sink)
 			exp := startMetricsExporter(t, test.baseURL, test.overrideURL)
 
@@ -187,8 +187,8 @@ func TestMetricsRoundTrip(t *testing.T) {
 func TestLogsError(t *testing.T) {
 	addr := testutil.GetAvailableLocalAddress(t)
 
-	sink := new(exportertest.SinkLogsExporter)
-	sink.SetConsumeLogError(errors.New("my_error"))
+	sink := new(consumertest.SinkLogs)
+	sink.SetConsumeError(errors.New("my_error"))
 	startLogsReceiver(t, addr, sink)
 	exp := startLogsExporter(t, "", fmt.Sprintf("http://%s/v1/logs", addr))
 
@@ -223,7 +223,7 @@ func TestLogsRoundTrip(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			sink := new(exportertest.SinkLogsExporter)
+			sink := new(consumertest.SinkLogs)
 			startLogsReceiver(t, addr, sink)
 			exp := startLogsExporter(t, test.baseURL, test.overrideURL)