[consumer] Add tracing to all the components

.chloggen/add-tracing-information-to-all-components.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: consumer
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Add tracing information to all the components"
+
+# One or more tracking issues or pull requests related to the change
+issues: [8804]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

consumer/consumer.go

@@ -12,6 +12,9 @@ import (
 // Capabilities describes the capabilities of a Processor.
 type Capabilities = internal.Capabilities
 
+// ObsReport describes the observability report of a consumer.
+type ObsReport = internal.ObsReport
+
 var errNilFunc = errors.New("nil consumer func")
 
 // Option to construct new consumers.
@@ -24,3 +27,9 @@ func WithCapabilities(capabilities Capabilities) Option {
 		o.Cap = capabilities
 	}
 }
+
+func WithObsReport(report ObsReport) Option {
+	return func(o *internal.BaseImpl) {
+		o.ObsReport = report
+	}
+}

consumer/internal/consumer.go

@@ -18,7 +18,8 @@ type BaseConsumer interface {
 }
 
 type BaseImpl struct {
-	Cap Capabilities
+	Cap       Capabilities
+	ObsReport ObsReport
 }
 
 // Option to construct new consumers.
@@ -31,7 +32,8 @@ func (bs BaseImpl) Capabilities() Capabilities {
 
 func NewBaseImpl(options ...Option) *BaseImpl {
 	bs := &BaseImpl{
-		Cap: Capabilities{MutatesData: false},
+		Cap:       Capabilities{MutatesData: false},
+		ObsReport: noopObsReport,
 	}
 
 	for _, op := range options {

consumer/internal/obsreport.go

@@ -0,0 +1,23 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "go.opentelemetry.io/collector/consumer/internal"
+
+import "context"
+
+// ObsReport contains information required to make an implementor
+// of Consumer observable.
+type ObsReport interface {
+	StartTracesOp(context.Context) context.Context
+	EndTracesOp(context.Context, int, error)
+}
+
+type baseObsReport struct{}
+
+func (bor baseObsReport) StartTracesOp(ctx context.Context) context.Context {
+	return ctx
+}
+
+func (bor baseObsReport) EndTracesOp(_ context.Context, _ int, _ error) {}
+
+var noopObsReport = baseObsReport{}

consumer/logs.go

@@ -36,8 +36,17 @@ func NewLogs(consume ConsumeLogsFunc, options ...Option) (Logs, error) {
 	if consume == nil {
 		return nil, errNilFunc
 	}
+	baseImpl := internal.NewBaseImpl(options...)
+	fn := func(ctx context.Context, ld plog.Logs) error {
+		baseImpl.ObsReport.StartTracesOp(ctx)
+		logRecordCount := ld.LogRecordCount()
+		err := consume(ctx, ld)
+		baseImpl.ObsReport.EndTracesOp(ctx, logRecordCount, err)
+		return err
+	}
+
 	return &baseLogs{
-		BaseImpl:        internal.NewBaseImpl(options...),
-		ConsumeLogsFunc: consume,
+		BaseImpl:        baseImpl,
+		ConsumeLogsFunc: fn,
 	}, nil
 }

consumer/logs_test.go

@@ -13,6 +13,20 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 )
 
+type mockObsReport struct {
+	StartTracesOpCalled int
+	EndTracesOpCalled   int
+}
+
+func (m *mockObsReport) StartTracesOp(ctx context.Context) context.Context {
+	m.StartTracesOpCalled++
+	return ctx
+}
+
+func (m *mockObsReport) EndTracesOp(_ context.Context, _ int, _ error) {
+	m.EndTracesOpCalled++
+}
+
 func TestDefaultLogs(t *testing.T) {
 	cp, err := NewLogs(func(context.Context, plog.Logs) error { return nil })
 	assert.NoError(t, err)
@@ -34,6 +48,17 @@ func TestWithCapabilitiesLogs(t *testing.T) {
 	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
+func TestWithObsReportLogs(t *testing.T) {
+	obsr := &mockObsReport{}
+	cp, err := NewLogs(
+		func(context.Context, plog.Logs) error { return nil },
+		WithObsReport(obsr))
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeLogs(context.Background(), plog.NewLogs()))
+	assert.Equal(t, 1, obsr.StartTracesOpCalled)
+	assert.Equal(t, 1, obsr.EndTracesOpCalled)
+}
+
 func TestConsumeLogs(t *testing.T) {
 	consumeCalled := false
 	cp, err := NewLogs(func(context.Context, plog.Logs) error { consumeCalled = true; return nil })

consumer/metrics.go

@@ -36,8 +36,17 @@ func NewMetrics(consume ConsumeMetricsFunc, options ...Option) (Metrics, error)
 	if consume == nil {
 		return nil, errNilFunc
 	}
+	baseImpl := internal.NewBaseImpl(options...)
+	fn := func(ctx context.Context, ld pmetric.Metrics) error {
+		ctx = baseImpl.ObsReport.StartTracesOp(ctx)
+		dataPointCount := ld.DataPointCount()
+		err := consume(ctx, ld)
+		baseImpl.ObsReport.EndTracesOp(ctx, dataPointCount, err)
+		return err
+	}
+
 	return &baseMetrics{
-		BaseImpl:           internal.NewBaseImpl(options...),
-		ConsumeMetricsFunc: consume,
+		BaseImpl:           baseImpl,
+		ConsumeMetricsFunc: fn,
 	}, nil
 }

consumer/metrics_test.go

@@ -34,6 +34,17 @@ func TestWithCapabilitiesMetrics(t *testing.T) {
 	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
+func TestWithObsReportMetrics(t *testing.T) {
+	obsr := &mockObsReport{}
+	cp, err := NewMetrics(
+		func(context.Context, pmetric.Metrics) error { return nil },
+		WithObsReport(obsr))
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeMetrics(context.Background(), pmetric.NewMetrics()))
+	assert.Equal(t, 1, obsr.StartTracesOpCalled)
+	assert.Equal(t, 1, obsr.EndTracesOpCalled)
+}
+
 func TestConsumeMetrics(t *testing.T) {
 	consumeCalled := false
 	cp, err := NewMetrics(func(context.Context, pmetric.Metrics) error { consumeCalled = true; return nil })

consumer/traces.go

@@ -36,8 +36,18 @@ func NewTraces(consume ConsumeTracesFunc, options ...Option) (Traces, error) {
 	if consume == nil {
 		return nil, errNilFunc
 	}
+
+	baseImpl := internal.NewBaseImpl(options...)
+	fn := func(ctx context.Context, td ptrace.Traces) error {
+		ctx = baseImpl.ObsReport.StartTracesOp(ctx)
+		spanCount := td.SpanCount()
+		err := consume(ctx, td)
+		baseImpl.ObsReport.EndTracesOp(ctx, spanCount, err)
+		return err
+	}
+
 	return &baseTraces{
-		BaseImpl:          internal.NewBaseImpl(options...),
-		ConsumeTracesFunc: consume,
+		BaseImpl:          baseImpl,
+		ConsumeTracesFunc: fn,
 	}, nil
 }

consumer/traces_test.go

@@ -34,6 +34,17 @@ func TestWithCapabilitiesTraces(t *testing.T) {
 	assert.Equal(t, Capabilities{MutatesData: true}, cp.Capabilities())
 }
 
+func TestWithWithObsReportTraces(t *testing.T) {
+	obsr := &mockObsReport{}
+	cp, err := NewTraces(
+		func(context.Context, ptrace.Traces) error { return nil },
+		WithObsReport(obsr))
+	assert.NoError(t, err)
+	assert.NoError(t, cp.ConsumeTraces(context.Background(), ptrace.NewTraces()))
+	assert.Equal(t, 1, obsr.StartTracesOpCalled)
+	assert.Equal(t, 1, obsr.EndTracesOpCalled)
+}
+
 func TestConsumeTraces(t *testing.T) {
 	consumeCalled := false
 	cp, err := NewTraces(func(context.Context, ptrace.Traces) error { consumeCalled = true; return nil })