[chore] Create service/internal/obsconsumer package

Author: djaglowski

service/internal/obsconsumer/logs.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package obsconsumer // import "go.opentelemetry.io/collector/service/internal/obsconsumer"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+var _ consumer.Logs = Logs{}
+
+func NewLogs(consumer consumer.Logs, itemCounter metric.Int64Counter, opts ...Option) Logs {
+	o := options{}
+	for _, opt := range opts {
+		opt.apply(&o)
+	}
+	return Logs{
+		consumer:    consumer,
+		itemCounter: itemCounter,
+		options:     o,
+	}
+}
+
+type Logs struct {
+	consumer    consumer.Logs
+	itemCounter metric.Int64Counter
+	options
+}
+
+func (c Logs) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	var attrs []attribute.KeyValue
+	attrs = append(attrs, c.options.staticDataPointAttributes...)
+	err := c.consumer.ConsumeLogs(ctx, ld)
+	outcome := "success"
+	if err != nil {
+		outcome = "failure"
+	}
+	attrs = append(attrs, attribute.String("outcome", outcome))
+	c.itemCounter.Add(ctx, int64(ld.ResourceLogs().Len()), metric.WithAttributes(attrs...))
+	return err
+}
+
+func (c Logs) Capabilities() consumer.Capabilities {
+	return c.consumer.Capabilities()
+}

service/internal/obsconsumer/logs_test.go

@@ -0,0 +1,229 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package obsconsumer_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+
+	"go.opentelemetry.io/collector/service/internal/obsconsumer"
+)
+
+type mockLogsConsumer struct {
+	err error
+}
+
+func (m *mockLogsConsumer) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	return m.err
+}
+
+func (m *mockLogsConsumer) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func TestLogsConsumeSuccess(t *testing.T) {
+	ctx := context.Background()
+	mockConsumer := &mockLogsConsumer{}
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	meter := mp.Meter("test")
+	counter, err := meter.Int64Counter("test_counter")
+	require.NoError(t, err)
+
+	consumer := obsconsumer.NewLogs(mockConsumer, counter)
+
+	ld := plog.NewLogs()
+	rl := ld.ResourceLogs().AppendEmpty()
+	rl.ScopeLogs().AppendEmpty()
+
+	err = consumer.ConsumeLogs(ctx, ld)
+	require.NoError(t, err)
+
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	require.Len(t, rm.ScopeMetrics[0].Metrics, 1)
+
+	metric := rm.ScopeMetrics[0].Metrics[0]
+	require.Equal(t, "test_counter", metric.Name)
+
+	data := metric.Data.(metricdata.Sum[int64])
+	require.Len(t, data.DataPoints, 1)
+	require.Equal(t, int64(1), data.DataPoints[0].Value)
+
+	attrs := data.DataPoints[0].Attributes
+	require.Equal(t, 1, attrs.Len())
+	val, ok := attrs.Value(attribute.Key("outcome"))
+	require.True(t, ok)
+	require.Equal(t, "success", val.Emit())
+}
+
+func TestLogsConsumeFailure(t *testing.T) {
+	ctx := context.Background()
+	expectedErr := errors.New("test error")
+	mockConsumer := &mockLogsConsumer{err: expectedErr}
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	meter := mp.Meter("test")
+	counter, err := meter.Int64Counter("test_counter")
+	require.NoError(t, err)
+
+	consumer := obsconsumer.NewLogs(mockConsumer, counter)
+
+	ld := plog.NewLogs()
+	rl := ld.ResourceLogs().AppendEmpty()
+	rl.ScopeLogs().AppendEmpty()
+
+	err = consumer.ConsumeLogs(ctx, ld)
+	assert.Equal(t, expectedErr, err)
+
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	require.Len(t, rm.ScopeMetrics[0].Metrics, 1)
+
+	metric := rm.ScopeMetrics[0].Metrics[0]
+	require.Equal(t, "test_counter", metric.Name)
+
+	data := metric.Data.(metricdata.Sum[int64])
+	require.Len(t, data.DataPoints, 1)
+	require.Equal(t, int64(1), data.DataPoints[0].Value)
+
+	attrs := data.DataPoints[0].Attributes
+	require.Equal(t, 1, attrs.Len())
+	val, ok := attrs.Value(attribute.Key("outcome"))
+	require.True(t, ok)
+	require.Equal(t, "failure", val.Emit())
+}
+
+func TestLogsWithStaticAttributes(t *testing.T) {
+	ctx := context.Background()
+	mockConsumer := &mockLogsConsumer{}
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	meter := mp.Meter("test")
+	counter, err := meter.Int64Counter("test_counter")
+	require.NoError(t, err)
+
+	staticAttr := attribute.String("test", "value")
+	consumer := obsconsumer.NewLogs(mockConsumer, counter, obsconsumer.WithStaticDataPointAttribute(staticAttr))
+
+	ld := plog.NewLogs()
+	rl := ld.ResourceLogs().AppendEmpty()
+	rl.ScopeLogs().AppendEmpty()
+
+	err = consumer.ConsumeLogs(ctx, ld)
+	require.NoError(t, err)
+
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	require.Len(t, rm.ScopeMetrics[0].Metrics, 1)
+
+	metric := rm.ScopeMetrics[0].Metrics[0]
+	require.Equal(t, "test_counter", metric.Name)
+
+	data := metric.Data.(metricdata.Sum[int64])
+	require.Len(t, data.DataPoints, 1)
+	require.Equal(t, int64(1), data.DataPoints[0].Value)
+
+	attrs := data.DataPoints[0].Attributes
+	require.Equal(t, 2, attrs.Len())
+	val, ok := attrs.Value(attribute.Key("test"))
+	require.True(t, ok)
+	require.Equal(t, "value", val.Emit())
+	val, ok = attrs.Value(attribute.Key("outcome"))
+	require.True(t, ok)
+	require.Equal(t, "success", val.Emit())
+}
+
+func TestLogsMultipleItemsMixedOutcomes(t *testing.T) {
+	ctx := context.Background()
+	expectedErr := errors.New("test error")
+	mockConsumer := &mockLogsConsumer{}
+
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	meter := mp.Meter("test")
+	counter, err := meter.Int64Counter("test_counter")
+	require.NoError(t, err)
+
+	consumer := obsconsumer.NewLogs(mockConsumer, counter)
+
+	// First batch: 2 successful items
+	ld1 := plog.NewLogs()
+	for i := 0; i < 2; i++ {
+		rl := ld1.ResourceLogs().AppendEmpty()
+		rl.ScopeLogs().AppendEmpty()
+	}
+	err = consumer.ConsumeLogs(ctx, ld1)
+	require.NoError(t, err)
+
+	// Second batch: 1 failed item
+	mockConsumer.err = expectedErr
+	ld2 := plog.NewLogs()
+	rl := ld2.ResourceLogs().AppendEmpty()
+	rl.ScopeLogs().AppendEmpty()
+	err = consumer.ConsumeLogs(ctx, ld2)
+	assert.Equal(t, expectedErr, err)
+
+	// Third batch: 2 successful items
+	mockConsumer.err = nil
+	ld3 := plog.NewLogs()
+	for i := 0; i < 2; i++ {
+		rl := ld3.ResourceLogs().AppendEmpty()
+		rl.ScopeLogs().AppendEmpty()
+	}
+	err = consumer.ConsumeLogs(ctx, ld3)
+	require.NoError(t, err)
+
+	// Fourth batch: 1 failed item
+	mockConsumer.err = expectedErr
+	ld4 := plog.NewLogs()
+	rl = ld4.ResourceLogs().AppendEmpty()
+	rl.ScopeLogs().AppendEmpty()
+	err = consumer.ConsumeLogs(ctx, ld4)
+	assert.Equal(t, expectedErr, err)
+
+	var rm metricdata.ResourceMetrics
+	err = reader.Collect(ctx, &rm)
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	require.Len(t, rm.ScopeMetrics[0].Metrics, 1)
+
+	metric := rm.ScopeMetrics[0].Metrics[0]
+	require.Equal(t, "test_counter", metric.Name)
+
+	data := metric.Data.(metricdata.Sum[int64])
+	require.Len(t, data.DataPoints, 2)
+
+	// Find success and failure data points
+	var successDP, failureDP metricdata.DataPoint[int64]
+	for _, dp := range data.DataPoints {
+		val, ok := dp.Attributes.Value(attribute.Key("outcome"))
+		if ok && val.Emit() == "success" {
+			successDP = dp
+		} else {
+			failureDP = dp
+		}
+	}
+
+	require.Equal(t, int64(4), successDP.Value)
+	require.Equal(t, int64(2), failureDP.Value)
+}

service/internal/obsconsumer/metrics.go

@@ -0,0 +1,51 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package obsconsumer // import "go.opentelemetry.io/collector/service/internal/obsconsumer"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+var _ consumer.Metrics = Metrics{}
+
+func NewMetrics(consumer consumer.Metrics, itemCounter metric.Int64Counter, opts ...Option) Metrics {
+	o := options{}
+	for _, opt := range opts {
+		opt.apply(&o)
+	}
+	return Metrics{
+		consumer:    consumer,
+		itemCounter: itemCounter,
+		options:     o,
+	}
+}
+
+type Metrics struct {
+	consumer    consumer.Metrics
+	itemCounter metric.Int64Counter
+	options
+}
+
+func (c Metrics) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
+	var attrs []attribute.KeyValue
+	attrs = append(attrs, c.options.staticDataPointAttributes...)
+	err := c.consumer.ConsumeMetrics(ctx, md)
+	outcome := "success"
+	if err != nil {
+		outcome = "failure"
+	}
+	attrs = append(attrs, attribute.String("outcome", outcome))
+	c.itemCounter.Add(ctx, int64(md.ResourceMetrics().Len()), metric.WithAttributes(attrs...))
+	return err
+}
+
+func (c Metrics) Capabilities() consumer.Capabilities {
+	return c.consumer.Capabilities()
+}