Add initial implementation of the consumerhelper (open-telemetry#3146)

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

consumer/consumerhelper/common.go

@@ -0,0 +1,55 @@
+// 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 consumerhelper
+
+import (
+	"errors"
+
+	"go.opentelemetry.io/collector/consumer"
+)
+
+var errNilFunc = errors.New("nil consumer func")
+
+type baseConsumer struct {
+	capabilities consumer.Capabilities
+}
+
+// Option apply changes to internalOptions.
+type Option func(*baseConsumer)
+
+// WithCapabilities overrides the default GetCapabilities function for an processor.
+// The default GetCapabilities function returns mutable capabilities.
+func WithCapabilities(capabilities consumer.Capabilities) Option {
+	return func(o *baseConsumer) {
+		o.capabilities = capabilities
+	}
+}
+
+// Capabilities implementation of the base Consumer.
+func (bs baseConsumer) Capabilities() consumer.Capabilities {
+	return bs.capabilities
+}
+
+func newBaseConsumer(options ...Option) *baseConsumer {
+	bs := &baseConsumer{
+		capabilities: consumer.Capabilities{MutatesData: false},
+	}
+
+	for _, op := range options {
+		op(bs)
+	}
+
+	return bs
+}

consumer/consumerhelper/common_test.go

@@ -0,0 +1,36 @@
+// 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 consumerhelper
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/consumer"
+)
+
+func TestDefaultOptions(t *testing.T) {
+	bp := newBaseConsumer()
+	assert.Equal(t, consumer.Capabilities{MutatesData: false}, bp.Capabilities())
+}
+
+func TestWithCapabilities(t *testing.T) {
+	bpMutate := newBaseConsumer(WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, bpMutate.Capabilities())
+
+	bpNotMutate := newBaseConsumer(WithCapabilities(consumer.Capabilities{MutatesData: false}))
+	assert.Equal(t, consumer.Capabilities{MutatesData: false}, bpNotMutate.Capabilities())
+}

consumer/consumerhelper/logs.go

@@ -0,0 +1,46 @@
+// 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 consumerhelper
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+// ConsumeLogsFunc is a helper function that is similar to ConsumeLogs.
+type ConsumeLogsFunc func(ctx context.Context, ld pdata.Logs) error
+
+// ConsumeLogs calls f(ctx, ld).
+func (f ConsumeLogsFunc) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
+	return f(ctx, ld)
+}
+
+type baseLogs struct {
+	*baseConsumer
+	ConsumeLogsFunc
+}
+
+// NewLogs returns a consumer.Logs configured with the provided options.
+func NewLogs(consume ConsumeLogsFunc, options ...Option) (consumer.Logs, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseLogs{
+		baseConsumer:    newBaseConsumer(options...),
+		ConsumeLogsFunc: consume,
+	}, nil
+}

consumer/consumerhelper/logs_test.go

@@ -0,0 +1,62 @@
+// 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 consumerhelper
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+func TestDefaultLogs(t *testing.T) {
+	cp, err := NewLogs(func(context.Context, pdata.Logs) error { return nil })
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
+	assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities())
+}
+
+func TestNilFuncLogs(t *testing.T) {
+	_, err := NewLogs(nil)
+	assert.Equal(t, errNilFunc, err)
+}
+
+func TestWithCapabilitiesLogs(t *testing.T) {
+	cp, err := NewLogs(
+		func(context.Context, pdata.Logs) error { return nil },
+		WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities())
+}
+
+func TestConsumeLogs(t *testing.T) {
+	consumeCalled := false
+	cp, err := NewLogs(func(context.Context, pdata.Logs) error { consumeCalled = true; return nil })
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
+	assert.True(t, consumeCalled)
+}
+
+func TestConsumeLogs_ReturnError(t *testing.T) {
+	want := errors.New("my_error")
+	cp, err := NewLogs(func(context.Context, pdata.Logs) error { return want })
+	assert.NoError(t, err)
+	assert.Equal(t, want, cp.ConsumeLogs(context.Background(), pdata.NewLogs()))
+}

consumer/consumerhelper/metrics.go

@@ -0,0 +1,46 @@
+// 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 consumerhelper
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+// ConsumeMetricsFunc is a helper function that is similar to ConsumeMetrics.
+type ConsumeMetricsFunc func(ctx context.Context, ld pdata.Metrics) error
+
+// ConsumeMetrics calls f(ctx, ld).
+func (f ConsumeMetricsFunc) ConsumeMetrics(ctx context.Context, ld pdata.Metrics) error {
+	return f(ctx, ld)
+}
+
+type baseMetrics struct {
+	*baseConsumer
+	ConsumeMetricsFunc
+}
+
+// NewMetrics returns a consumer.Metrics configured with the provided options.
+func NewMetrics(consume ConsumeMetricsFunc, options ...Option) (consumer.Metrics, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseMetrics{
+		baseConsumer:       newBaseConsumer(options...),
+		ConsumeMetricsFunc: consume,
+	}, nil
+}

consumer/consumerhelper/metrics_test.go

@@ -0,0 +1,62 @@
+// 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 consumerhelper
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+func TestDefaultMetrics(t *testing.T) {
+	cp, err := NewMetrics(func(context.Context, pdata.Metrics) error { return nil })
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
+	assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities())
+}
+
+func TestNilFuncMetrics(t *testing.T) {
+	_, err := NewMetrics(nil)
+	assert.Equal(t, errNilFunc, err)
+}
+
+func TestWithCapabilitiesMetrics(t *testing.T) {
+	cp, err := NewMetrics(
+		func(context.Context, pdata.Metrics) error { return nil },
+		WithCapabilities(consumer.Capabilities{MutatesData: true}))
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
+	assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities())
+}
+
+func TestConsumeMetrics(t *testing.T) {
+	consumeCalled := false
+	cp, err := NewMetrics(func(context.Context, pdata.Metrics) error { consumeCalled = true; return nil })
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
+	assert.True(t, consumeCalled)
+}
+
+func TestConsumeMetrics_ReturnError(t *testing.T) {
+	want := errors.New("my_error")
+	cp, err := NewMetrics(func(context.Context, pdata.Metrics) error { return want })
+	assert.NoError(t, err)
+	assert.Equal(t, want, cp.ConsumeMetrics(context.Background(), pdata.NewMetrics()))
+}

consumer/consumerhelper/traces.go

@@ -0,0 +1,46 @@
+// 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 consumerhelper
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+// ConsumeTracesFunc is a helper function that is similar to ConsumeTraces.
+type ConsumeTracesFunc func(ctx context.Context, ld pdata.Traces) error
+
+// ConsumeTraces calls f(ctx, ld).
+func (f ConsumeTracesFunc) ConsumeTraces(ctx context.Context, ld pdata.Traces) error {
+	return f(ctx, ld)
+}
+
+type baseTraces struct {
+	*baseConsumer
+	ConsumeTracesFunc
+}
+
+// NewTraces returns a consumer.Traces configured with the provided options.
+func NewTraces(consume ConsumeTracesFunc, options ...Option) (consumer.Traces, error) {
+	if consume == nil {
+		return nil, errNilFunc
+	}
+	return &baseTraces{
+		baseConsumer:      newBaseConsumer(options...),
+		ConsumeTracesFunc: consume,
+	}, nil
+}