[chore] add profiles support to processor receiver builder (#10955)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

This allows building profiles in the processor builder.
As this is only changing internal modules, I am marking it as chore (and
skipping changelog entry).

#### Link to tracking issue
#10375

cc @mx-psi

cmd/builder/internal/builder/main_test.go

@@ -77,6 +77,7 @@ var (
 		"/processor",
 		"/processor/batchprocessor",
 		"/processor/memorylimiterprocessor",
+		"/processor/processorprofiles",
 		"/receiver",
 		"/receiver/nopreceiver",
 		"/receiver/otlpreceiver",

cmd/otelcorecol/builder-config.yaml

@@ -84,11 +84,12 @@ replaces:
   - go.opentelemetry.io/collector/pdata/testdata => ../../pdata/testdata
   - go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
   - go.opentelemetry.io/collector/processor => ../../processor
+  - go.opentelemetry.io/collector/processor/batchprocessor => ../../processor/batchprocessor
+  - go.opentelemetry.io/collector/processor/memorylimiterprocessor => ../../processor/memorylimiterprocessor
+  - go.opentelemetry.io/collector/processor/processorprofiles => ../../processor/processorprofiles
   - go.opentelemetry.io/collector/receiver => ../../receiver
   - go.opentelemetry.io/collector/receiver/nopreceiver => ../../receiver/nopreceiver
   - go.opentelemetry.io/collector/receiver/otlpreceiver => ../../receiver/otlpreceiver
   - go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
-  - go.opentelemetry.io/collector/processor/batchprocessor => ../../processor/batchprocessor
-  - go.opentelemetry.io/collector/processor/memorylimiterprocessor => ../../processor/memorylimiterprocessor
   - go.opentelemetry.io/collector/semconv => ../../semconv
   - go.opentelemetry.io/collector/service => ../../service

cmd/otelcorecol/go.mod

@@ -103,6 +103,7 @@ require (
 	go.opentelemetry.io/collector/pdata v1.14.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/semconv v0.108.1 // indirect
 	go.opentelemetry.io/collector/service v0.108.1 // indirect
@@ -230,6 +231,12 @@ replace go.opentelemetry.io/collector/pdata/pprofile => ../../pdata/pprofile
 
 replace go.opentelemetry.io/collector/processor => ../../processor
 
+replace go.opentelemetry.io/collector/processor/batchprocessor => ../../processor/batchprocessor
+
+replace go.opentelemetry.io/collector/processor/memorylimiterprocessor => ../../processor/memorylimiterprocessor
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../../processor/processorprofiles
+
 replace go.opentelemetry.io/collector/receiver => ../../receiver
 
 replace go.opentelemetry.io/collector/receiver/nopreceiver => ../../receiver/nopreceiver
@@ -238,10 +245,6 @@ replace go.opentelemetry.io/collector/receiver/otlpreceiver => ../../receiver/ot
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
-replace go.opentelemetry.io/collector/processor/batchprocessor => ../../processor/batchprocessor
-
-replace go.opentelemetry.io/collector/processor/memorylimiterprocessor => ../../processor/memorylimiterprocessor
-
 replace go.opentelemetry.io/collector/semconv => ../../semconv
 
 replace go.opentelemetry.io/collector/service => ../../service

internal/e2e/go.mod

@@ -82,6 +82,7 @@ require (
 	go.opentelemetry.io/collector/internal/globalgates v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/collector/processor v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/semconv v0.108.1 // indirect
 	go.opentelemetry.io/contrib/config v0.9.0 // indirect
@@ -191,3 +192,5 @@ replace go.opentelemetry.io/collector/component/componentprofiles => ../../compo
 replace go.opentelemetry.io/collector/internal/globalgates => ../globalgates
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../../processor/processorprofiles

otelcol/go.mod

@@ -71,6 +71,7 @@ require (
 	go.opentelemetry.io/collector/pdata v1.14.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/semconv v0.108.1 // indirect
 	go.opentelemetry.io/contrib/config v0.9.0 // indirect
@@ -164,3 +165,5 @@ replace go.opentelemetry.io/collector/client => ../client
 replace go.opentelemetry.io/collector/component/componentstatus => ../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../processor/processorprofiles

otelcol/otelcoltest/go.mod

@@ -70,6 +70,7 @@ require (
 	go.opentelemetry.io/collector/pdata v1.14.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/semconv v0.108.1 // indirect
 	go.opentelemetry.io/contrib/config v0.9.0 // indirect
@@ -179,3 +180,5 @@ replace go.opentelemetry.io/collector/client => ../../client
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../../processor/processorprofiles

processor/batchprocessor/go.mod

@@ -49,6 +49,7 @@ require (
 	go.opentelemetry.io/collector/component/componentstatus v0.108.1 // indirect
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.51.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.29.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
@@ -93,3 +94,5 @@ replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consume
 replace go.opentelemetry.io/collector/consumer/consumertest => ../../consumer/consumertest
 
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../processorprofiles

processor/go.mod

@@ -13,7 +13,9 @@ require (
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.108.1
 	go.opentelemetry.io/collector/consumer/consumertest v0.108.1
 	go.opentelemetry.io/collector/pdata v1.14.1
+	go.opentelemetry.io/collector/pdata/pprofile v0.108.1
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1
 	go.opentelemetry.io/otel v1.29.0
 	go.opentelemetry.io/otel/metric v1.29.0
 	go.opentelemetry.io/otel/sdk/metric v1.29.0
@@ -39,7 +41,6 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.56.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
-	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.51.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.29.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
@@ -75,3 +76,5 @@ replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../consumer/c
 replace go.opentelemetry.io/collector/consumer/consumertest => ../consumer/consumertest
 
 replace go.opentelemetry.io/collector/component/componentstatus => ../component/componentstatus
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ./processorprofiles

processor/memorylimiterprocessor/go.mod

@@ -50,6 +50,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.108.1 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1 // indirect
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1 // indirect
 	go.opentelemetry.io/otel v1.29.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.51.0 // indirect
 	go.opentelemetry.io/otel/metric v1.29.0 // indirect
@@ -97,3 +98,5 @@ replace go.opentelemetry.io/collector/consumer/consumerprofiles => ../../consume
 replace go.opentelemetry.io/collector/consumer/consumertest => ../../consumer/consumertest
 
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../processorprofiles

processor/processortest/nop_processor.go

@@ -11,8 +11,10 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumerprofiles"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorprofiles"
 )
 
 var nopType = component.MustNewType("nop")
@@ -34,6 +36,7 @@ func NewNopFactory() processor.Factory {
 		processor.WithTraces(createTracesProcessor, component.StabilityLevelStable),
 		processor.WithMetrics(createMetricsProcessor, component.StabilityLevelStable),
 		processor.WithLogs(createLogsProcessor, component.StabilityLevelStable),
+		processorprofiles.WithProfiles(createProfilesProcessor, component.StabilityLevelAlpha),
 	)
 }
 
@@ -49,6 +52,10 @@ func createLogsProcessor(context.Context, processor.Settings, component.Config,
 	return nopInstance, nil
 }
 
+func createProfilesProcessor(context.Context, processor.Settings, component.Config, consumerprofiles.Profiles) (processorprofiles.Profiles, error) {
+	return nopInstance, nil
+}
+
 type nopConfig struct{}
 
 var nopInstance = &nopProcessor{

processor/processortest/nop_processor_test.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 )
 
@@ -46,6 +47,13 @@ func TestNewNopFactory(t *testing.T) {
 	assert.NoError(t, logs.Start(context.Background(), componenttest.NewNopHost()))
 	assert.NoError(t, logs.ConsumeLogs(context.Background(), plog.NewLogs()))
 	assert.NoError(t, logs.Shutdown(context.Background()))
+
+	profiles, err := factory.CreateProfilesProcessor(context.Background(), NewNopSettings(), cfg, consumertest.NewNop())
+	require.NoError(t, err)
+	assert.Equal(t, consumer.Capabilities{MutatesData: false}, profiles.Capabilities())
+	assert.NoError(t, profiles.Start(context.Background(), componenttest.NewNopHost()))
+	assert.NoError(t, profiles.ConsumeProfiles(context.Background(), pprofile.NewProfiles()))
+	assert.NoError(t, profiles.Shutdown(context.Background()))
 }
 
 func TestNewNopBuilder(t *testing.T) {

service/go.mod

@@ -27,6 +27,7 @@ require (
 	go.opentelemetry.io/collector/pdata v1.14.1
 	go.opentelemetry.io/collector/pdata/testdata v0.108.1
 	go.opentelemetry.io/collector/processor v0.108.1
+	go.opentelemetry.io/collector/processor/processorprofiles v0.108.1
 	go.opentelemetry.io/collector/receiver v0.108.1
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.108.1
 	go.opentelemetry.io/collector/semconv v0.108.1
@@ -172,3 +173,5 @@ replace go.opentelemetry.io/collector/client => ../client
 replace go.opentelemetry.io/collector/internal/globalgates => ../internal/globalgates
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/processor/processorprofiles => ../processor/processorprofiles

service/internal/builders/processor.go

@@ -9,7 +9,9 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumerprofiles"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/collector/processor/processorprofiles"
 	"go.opentelemetry.io/collector/processor/processortest"
 )
 
@@ -91,6 +93,25 @@ func (b *ProcessorBuilder) CreateLogs(ctx context.Context, set processor.Setting
 	return f.CreateLogsProcessor(ctx, set, cfg, next)
 }
 
+// CreateProfiles creates a Profiles processor based on the settings and config.
+func (b *ProcessorBuilder) CreateProfiles(ctx context.Context, set processor.Settings, next consumerprofiles.Profiles) (processorprofiles.Profiles, error) {
+	if next == nil {
+		return nil, errNilNextConsumer
+	}
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("processor %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("processor factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.ProfilesProcessorStability())
+	return f.CreateProfilesProcessor(ctx, set, cfg, next)
+}
+
 func (b *ProcessorBuilder) Factory(componentType component.Type) component.Factory {
 	return b.factories[componentType]
 }