Support profiles in consumertest

.chloggen/profiles-consumertest.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/consumertest
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Allow testing profiles with consumertest.
+
+# One or more tracking issues or pull requests related to the change
+issues: [10692]
+
+# (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: [api]

Makefile

@@ -278,6 +278,7 @@ check-contrib:
  -replace go.opentelemetry.io/collector/connector=$(CURDIR)/connector \
  -replace go.opentelemetry.io/collector/connector/forwardconnector=$(CURDIR)/connector/forwardconnector \
  -replace go.opentelemetry.io/collector/consumer=$(CURDIR)/consumer \
+ -replace go.opentelemetry.io/collector/consumerprofiles=$(CURDIR)/consumerprofiles \
  -replace go.opentelemetry.io/collector/exporter=$(CURDIR)/exporter \
  -replace go.opentelemetry.io/collector/exporter/debugexporter=$(CURDIR)/exporter/debugexporter \
  -replace go.opentelemetry.io/collector/exporter/loggingexporter=$(CURDIR)/exporter/loggingexporter \
@@ -337,6 +338,7 @@ restore-contrib:
  -dropreplace go.opentelemetry.io/collector/connector \
  -dropreplace go.opentelemetry.io/collector/connector/forwardconnector \
  -dropreplace go.opentelemetry.io/collector/consumer \
+ -dropreplace go.opentelemetry.io/collector/consumerprofiles \
  -dropreplace go.opentelemetry.io/collector/exporter \
  -dropreplace go.opentelemetry.io/collector/exporter/debugexporter \
  -dropreplace go.opentelemetry.io/collector/exporter/loggingexporter \

cmd/builder/internal/builder/main_test.go

@@ -59,6 +59,7 @@ var (
  "/confmap/provider/httpsprovider",
  "/confmap/provider/yamlprovider",
  "/consumer",
+ "/consumer/consumerprofiles",
  "/connector",
  "/exporter",
  "/exporter/debugexporter",

cmd/mdatagen/go.mod

@@ -46,8 +46,10 @@ require (
  github.com/prometheus/client_model v0.6.1 // indirect
  github.com/prometheus/common v0.55.0 // indirect
  github.com/prometheus/procfs v0.15.1 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
+ go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.opentelemetry.io/otel v1.28.0 // indirect
  go.opentelemetry.io/otel/exporters/prometheus v0.50.0 // indirect
  go.opentelemetry.io/otel/sdk v1.28.0 // indirect
@@ -93,3 +95,5 @@ retract (
 replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
 
 replace go.opentelemetry.io/collector/internal/globalgates => ../../internal/globalgates
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

cmd/otelcorecol/builder-config.yaml

@@ -62,6 +62,7 @@ replaces:
  - go.opentelemetry.io/collector/confmap/provider/httpsprovider => ../../confmap/provider/httpsprovider
  - go.opentelemetry.io/collector/confmap/provider/yamlprovider => ../../confmap/provider/yamlprovider
  - go.opentelemetry.io/collector/consumer => ../../consumer
+ - go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles
  - go.opentelemetry.io/collector/connector => ../../connector
  - go.opentelemetry.io/collector/connector/forwardconnector => ../../connector/forwardconnector
  - go.opentelemetry.io/collector/exporter => ../../exporter

cmd/otelcorecol/go.mod

@@ -92,10 +92,12 @@ require (
  go.opentelemetry.io/collector/config/configtls v1.12.0 // indirect
  go.opentelemetry.io/collector/config/internal v0.105.0 // indirect
  go.opentelemetry.io/collector/consumer v0.105.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/extension/auth v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
  go.opentelemetry.io/collector/pdata v1.12.0 // indirect
+ go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.opentelemetry.io/collector/semconv v0.105.0 // indirect
  go.opentelemetry.io/collector/service v0.105.0 // indirect
  go.opentelemetry.io/contrib/config v0.8.0 // indirect
@@ -178,6 +180,8 @@ replace go.opentelemetry.io/collector/confmap/provider/yamlprovider => ../../con
 
 replace go.opentelemetry.io/collector/consumer => ../../consumer
 
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles
+
 replace go.opentelemetry.io/collector/connector => ../../connector
 
 replace go.opentelemetry.io/collector/connector/forwardconnector => ../../connector/forwardconnector

config/configgrpc/go.mod

@@ -105,3 +105,5 @@ replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
 replace go.opentelemetry.io/collector/component => ../../component
 
 replace go.opentelemetry.io/collector/consumer => ../../consumer
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

config/confighttp/go.mod

@@ -98,3 +98,5 @@ replace go.opentelemetry.io/collector/consumer => ../../consumer
 replace go.opentelemetry.io/collector/pdata/testdata => ../../pdata/testdata
 
 replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

config/internal/go.mod

@@ -37,3 +37,5 @@ replace go.opentelemetry.io/collector/pdata/testdata => ../../pdata/testdata
 replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
 
 replace go.opentelemetry.io/collector/internal/globalgates => ../../internal/globalgates
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

connector/forwardconnector/go.mod

@@ -38,8 +38,10 @@ require (
  github.com/prometheus/procfs v0.15.1 // indirect
  go.opentelemetry.io/collector v0.105.0 // indirect
  go.opentelemetry.io/collector/config/configtelemetry v0.105.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
+ go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.opentelemetry.io/otel v1.28.0 // indirect
  go.opentelemetry.io/otel/exporters/prometheus v0.50.0 // indirect
  go.opentelemetry.io/otel/metric v1.28.0 // indirect
@@ -83,3 +85,5 @@ retract (
 replace go.opentelemetry.io/collector/config/configtelemetry => ../../config/configtelemetry
 
 replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

connector/go.mod

@@ -32,6 +32,7 @@ require (
  github.com/prometheus/common v0.55.0 // indirect
  github.com/prometheus/procfs v0.15.1 // indirect
  go.opentelemetry.io/collector/config/configtelemetry v0.105.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.opentelemetry.io/otel v1.28.0 // indirect
  go.opentelemetry.io/otel/exporters/prometheus v0.50.0 // indirect
@@ -67,3 +68,5 @@ replace go.opentelemetry.io/collector/pdata/testdata => ../pdata/testdata
 replace go.opentelemetry.io/collector/pdata/pprofile => ../pdata/pprofile
 
 replace go.opentelemetry.io/collector/internal/globalgates => ../internal/globalgates
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../consumer/consumerprofiles

consumer/consumertest/consumer.go

@@ -7,8 +7,10 @@ import (
  "context"
 
  "go.opentelemetry.io/collector/consumer"
+ "go.opentelemetry.io/collector/consumer/consumerprofiles"
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
@@ -29,12 +31,16 @@ type Consumer interface {
  // ConsumeLogs to implement the consumer.Logs.
  ConsumeLogs(context.Context, plog.Logs) error
 
+ // ConsumeProfiles to implement the consumerprofiles.Profiles.
+ ConsumeProfiles(context.Context, pprofile.Profiles) error
+
  unexported()
 }
 
 var _ consumer.Logs = (Consumer)(nil)
 var _ consumer.Metrics = (Consumer)(nil)
 var _ consumer.Traces = (Consumer)(nil)
+var _ consumerprofiles.Profiles = (Consumer)(nil)
 
 type nonMutatingConsumer struct{}
 
@@ -48,6 +54,7 @@ type baseConsumer struct {
  consumer.ConsumeTracesFunc
  consumer.ConsumeMetricsFunc
  consumer.ConsumeLogsFunc
+ consumerprofiles.ConsumeProfilesFunc
 }
 
 func (bc baseConsumer) unexported() {}

consumer/consumertest/err.go

@@ -7,14 +7,16 @@ import (
 
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
 // NewErr returns a Consumer that just drops all received data and returns the specified error to Consume* callers.
 func NewErr(err error) Consumer {
  return &baseConsumer{
- ConsumeTracesFunc: func(context.Context, ptrace.Traces) error { return err },
- ConsumeMetricsFunc: func(context.Context, pmetric.Metrics) error { return err },
- ConsumeLogsFunc: func(context.Context, plog.Logs) error { return err },
+ ConsumeTracesFunc: func(context.Context, ptrace.Traces) error { return err },
+ ConsumeMetricsFunc: func(context.Context, pmetric.Metrics) error { return err },
+ ConsumeLogsFunc: func(context.Context, plog.Logs) error { return err },
+ ConsumeProfilesFunc: func(context.Context, pprofile.Profiles) error { return err },
  }
 }

consumer/consumertest/err_test.go

@@ -13,6 +13,7 @@ import (
 
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
@@ -24,4 +25,5 @@ func TestErr(t *testing.T) {
  assert.Equal(t, err, ec.ConsumeLogs(context.Background(), plog.NewLogs()))
  assert.Equal(t, err, ec.ConsumeMetrics(context.Background(), pmetric.NewMetrics()))
  assert.Equal(t, err, ec.ConsumeTraces(context.Background(), ptrace.NewTraces()))
+ assert.Equal(t, err, ec.ConsumeProfiles(context.Background(), pprofile.NewProfiles()))
 }

consumer/consumertest/nop.go

@@ -8,14 +8,16 @@ import (
 
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
 // NewNop returns a Consumer that just drops all received data and returns no error.
 func NewNop() Consumer {
  return &baseConsumer{
- ConsumeTracesFunc: func(context.Context, ptrace.Traces) error { return nil },
- ConsumeMetricsFunc: func(context.Context, pmetric.Metrics) error { return nil },
- ConsumeLogsFunc: func(context.Context, plog.Logs) error { return nil },
+ ConsumeTracesFunc: func(context.Context, ptrace.Traces) error { return nil },
+ ConsumeMetricsFunc: func(context.Context, pmetric.Metrics) error { return nil },
+ ConsumeLogsFunc: func(context.Context, plog.Logs) error { return nil },
+ ConsumeProfilesFunc: func(context.Context, pprofile.Profiles) error { return nil },
  }
 }

consumer/consumertest/nop_test.go

@@ -12,6 +12,7 @@ import (
 
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
@@ -22,4 +23,5 @@ func TestNop(t *testing.T) {
  assert.NoError(t, nc.ConsumeLogs(context.Background(), plog.NewLogs()))
  assert.NoError(t, nc.ConsumeMetrics(context.Background(), pmetric.NewMetrics()))
  assert.NoError(t, nc.ConsumeTraces(context.Background(), ptrace.NewTraces()))
+ assert.NoError(t, nc.ConsumeProfiles(context.Background(), pprofile.NewProfiles()))
 }

consumer/consumertest/sink.go

@@ -8,8 +8,10 @@ import (
  "sync"
 
  "go.opentelemetry.io/collector/consumer"
+ "go.opentelemetry.io/collector/consumer/consumerprofiles"
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
 )
 
@@ -156,3 +158,41 @@ func (sle *LogsSink) Reset() {
  sle.logs = nil
  sle.logRecordCount = 0
 }
+
+// ProfilesSink is a consumerprofiles.Profiles that acts like a sink that
+// stores all profiles and allows querying them for testing.
+type ProfilesSink struct {
+ nonMutatingConsumer
+ mu sync.Mutex
+ profiles []pprofile.Profiles
+}
+
+var _ consumerprofiles.Profiles = (*ProfilesSink)(nil)
+
+// ConsumeProfiles stores profiles to this sink.
+func (ste *ProfilesSink) ConsumeProfiles(_ context.Context, td pprofile.Profiles) error {
+ ste.mu.Lock()
+ defer ste.mu.Unlock()
+
+ ste.profiles = append(ste.profiles, td)
+
+ return nil
+}
+
+// AllProfiles returns the profiles stored by this sink since last Reset.
+func (ste *ProfilesSink) AllProfiles() []pprofile.Profiles {
+ ste.mu.Lock()
+ defer ste.mu.Unlock()
+
+ copyProfiles := make([]pprofile.Profiles, len(ste.profiles))
+ copy(copyProfiles, ste.profiles)
+ return copyProfiles
+}
+
+// Reset deletes any stored data.
+func (ste *ProfilesSink) Reset() {
+ ste.mu.Lock()
+ defer ste.mu.Unlock()
+
+ ste.profiles = nil
+}

consumer/consumertest/sink_test.go

@@ -12,6 +12,7 @@ import (
 
  "go.opentelemetry.io/collector/pdata/plog"
  "go.opentelemetry.io/collector/pdata/pmetric"
+ "go.opentelemetry.io/collector/pdata/pprofile"
  "go.opentelemetry.io/collector/pdata/ptrace"
  "go.opentelemetry.io/collector/pdata/testdata"
 )
@@ -60,3 +61,16 @@ func TestLogsSink(t *testing.T) {
  assert.Equal(t, 0, len(sink.AllLogs()))
  assert.Equal(t, 0, sink.LogRecordCount())
 }
+
+func TestProfilesSink(t *testing.T) {
+ sink := new(ProfilesSink)
+ td := testdata.GenerateProfiles(1)
+ want := make([]pprofile.Profiles, 0, 7)
+ for i := 0; i < 7; i++ {
+ require.NoError(t, sink.ConsumeProfiles(context.Background(), td))
+ want = append(want, td)
+ }
+ assert.Equal(t, want, sink.AllProfiles())
+ sink.Reset()
+ assert.Equal(t, 0, len(sink.AllProfiles()))
+}

consumer/go.mod

@@ -4,7 +4,9 @@ go 1.21.0
 
 require (
  github.com/stretchr/testify v1.9.0
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0
  go.opentelemetry.io/collector/pdata v1.12.0
+ go.opentelemetry.io/collector/pdata/pprofile v0.105.0
  go.opentelemetry.io/collector/pdata/testdata v0.105.0
  go.uber.org/goleak v1.3.0
 )
@@ -16,7 +18,6 @@ require (
  github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
  github.com/modern-go/reflect2 v1.0.2 // indirect
  github.com/pmezard/go-difflib v1.0.0 // indirect
- go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.uber.org/multierr v1.11.0 // indirect
  golang.org/x/net v0.25.0 // indirect
  golang.org/x/sys v0.20.0 // indirect
@@ -37,3 +38,5 @@ retract (
  v0.76.0 // Depends on retracted pdata v1.0.0-rc10 module, use v0.76.1
  v0.69.0 // Release failed, use v0.69.1
 )
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ./consumerprofiles

exporter/debugexporter/go.mod

@@ -42,6 +42,7 @@ require (
  github.com/prometheus/procfs v0.15.1 // indirect
  go.opentelemetry.io/collector v0.105.0 // indirect
  go.opentelemetry.io/collector/config/configretry v1.12.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/extension v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
@@ -90,3 +91,5 @@ replace go.opentelemetry.io/collector/extension => ../../extension
 replace go.opentelemetry.io/collector/config/configtelemetry => ../../config/configtelemetry
 
 replace go.opentelemetry.io/collector/config/configretry => ../../config/configretry
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles

exporter/go.mod

@@ -51,6 +51,7 @@ require (
  github.com/prometheus/common v0.55.0 // indirect
  github.com/prometheus/procfs v0.15.1 // indirect
  go.opentelemetry.io/collector/confmap v0.105.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
  go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
@@ -89,3 +90,5 @@ retract v0.76.0 // Depends on retracted pdata v1.0.0-rc10 module
 replace go.opentelemetry.io/collector/config/configretry => ../config/configretry
 
 replace go.opentelemetry.io/collector/config/configtelemetry => ../config/configtelemetry
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../consumer/consumerprofiles

exporter/loggingexporter/go.mod

@@ -42,9 +42,11 @@ require (
  go.opentelemetry.io/collector v0.105.0 // indirect
  go.opentelemetry.io/collector/config/configretry v1.12.0 // indirect
  go.opentelemetry.io/collector/consumer v0.105.0 // indirect
+ go.opentelemetry.io/collector/consumer/consumerprofiles v0.105.0 // indirect
  go.opentelemetry.io/collector/extension v0.105.0 // indirect
  go.opentelemetry.io/collector/featuregate v1.12.0 // indirect
  go.opentelemetry.io/collector/internal/globalgates v0.105.0 // indirect
+ go.opentelemetry.io/collector/pdata/pprofile v0.105.0 // indirect
  go.opentelemetry.io/collector/receiver v0.105.0 // indirect
  go.opentelemetry.io/otel v1.28.0 // indirect
  go.opentelemetry.io/otel/exporters/prometheus v0.50.0 // indirect
@@ -94,3 +96,5 @@ replace go.opentelemetry.io/collector/config/configtelemetry => ../../config/con
 replace go.opentelemetry.io/collector/config/configretry => ../../config/configretry
 
 replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
+
+replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consumer/consumerprofiles