[connector/forward]: Add support for Profiles to Profiles (#14092)

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

Adds support for Profiles to Profies in the forward connector.


<!--Describe what testing was performed and which tests were added.-->
#### Testing

Produce -> consume unit test

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Alex Boten <[email protected]>

Author: rogercoll

.chloggen/add_profiles_to_profiles_forward_connector.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: connector/forward
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for Profiles to Profiles
+
+# One or more tracking issues or pull requests related to the change
+issues: [14092]
+
+# (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: []

connector/forwardconnector/README.md

@@ -6,6 +6,7 @@
 | Distributions | [core], [contrib], [k8s] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector?query=is%3Aissue%20is%3Aopen%20label%3Aconnector%2Fforward%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector/issues?q=is%3Aopen+is%3Aissue+label%3Aconnector%2Fforward) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector?query=is%3Aissue%20is%3Aclosed%20label%3Aconnector%2Fforward%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector/issues?q=is%3Aclosed+is%3Aissue+label%3Aconnector%2Fforward) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
 [beta]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#beta
 [core]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
@@ -15,6 +16,7 @@
 
 | [Exporter Pipeline Type] | [Receiver Pipeline Type] | [Stability Level] |
 | ------------------------ | ------------------------ | ----------------- |
+| profiles | profiles | [development] |
 | traces | traces | [beta] |
 | metrics | metrics | [beta] |
 | logs | logs | [beta] |

connector/forwardconnector/forward.go

@@ -9,17 +9,20 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/connector/forwardconnector/internal/metadata"
+	"go.opentelemetry.io/collector/connector/xconnector"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/xconsumer"
 )
 
 // NewFactory returns a connector.Factory.
-func NewFactory() connector.Factory {
-	return connector.NewFactory(
+func NewFactory() xconnector.Factory {
+	return xconnector.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		connector.WithTracesToTraces(createTracesToTraces, metadata.TracesToTracesStability),
-		connector.WithMetricsToMetrics(createMetricsToMetrics, metadata.MetricsToMetricsStability),
-		connector.WithLogsToLogs(createLogsToLogs, metadata.LogsToLogsStability),
+		xconnector.WithTracesToTraces(createTracesToTraces, metadata.TracesToTracesStability),
+		xconnector.WithMetricsToMetrics(createMetricsToMetrics, metadata.MetricsToMetricsStability),
+		xconnector.WithLogsToLogs(createLogsToLogs, metadata.LogsToLogsStability),
+		xconnector.WithProfilesToProfiles(createProfilesToProfiles, metadata.ProfilesToProfilesStability),
 	)
 }
 
@@ -60,13 +63,24 @@ func createLogsToLogs(
 	return &forward{Logs: nextConsumer}, nil
 }
 
+// createProfilesToProfiles creates a profile receiver based on provided config.
+func createProfilesToProfiles(
+	_ context.Context,
+	_ connector.Settings,
+	_ component.Config,
+	nextConsumer xconsumer.Profiles,
+) (xconnector.Profiles, error) {
+	return &forward{Profiles: nextConsumer}, nil
+}
+
 // forward is used to pass signals directly from one pipeline to another.
 // This is useful when there is a need to replicate data and process it in more
 // than one way. It can also be used to join pipelines together.
 type forward struct {
 	consumer.Traces
 	consumer.Metrics
 	consumer.Logs
+	xconsumer.Profiles
 	component.StartFunc
 	component.ShutdownFunc
 }

connector/forwardconnector/forward_test.go

@@ -14,6 +14,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"
 )
 
@@ -41,9 +42,15 @@ func TestForward(t *testing.T) {
 	require.NoError(t, err)
 	assert.NotNil(t, logsToLogs)
 
+	profilesSink := new(consumertest.ProfilesSink)
+	profilesToProfiles, err := f.CreateProfilesToProfiles(ctx, set, cfg, profilesSink)
+	require.NoError(t, err)
+	assert.NotNil(t, profilesToProfiles)
+
 	assert.NoError(t, tracesToTraces.Start(ctx, host))
 	assert.NoError(t, metricsToMetrics.Start(ctx, host))
 	assert.NoError(t, logsToLogs.Start(ctx, host))
+	assert.NoError(t, profilesToProfiles.Start(ctx, host))
 
 	assert.NoError(t, tracesToTraces.ConsumeTraces(ctx, ptrace.NewTraces()))
 
@@ -54,11 +61,15 @@ func TestForward(t *testing.T) {
 	assert.NoError(t, logsToLogs.ConsumeLogs(ctx, plog.NewLogs()))
 	assert.NoError(t, logsToLogs.ConsumeLogs(ctx, plog.NewLogs()))
 
+	assert.NoError(t, profilesToProfiles.ConsumeProfiles(ctx, pprofile.NewProfiles()))
+
 	assert.NoError(t, tracesToTraces.Shutdown(ctx))
 	assert.NoError(t, metricsToMetrics.Shutdown(ctx))
 	assert.NoError(t, logsToLogs.Shutdown(ctx))
+	assert.NoError(t, profilesToProfiles.Shutdown(ctx))
 
 	assert.Len(t, tracesSink.AllTraces(), 1)
 	assert.Len(t, metricsSink.AllMetrics(), 2)
 	assert.Len(t, logsSink.AllLogs(), 3)
+	assert.Len(t, profilesSink.AllProfiles(), 1)
 }

connector/forwardconnector/go.mod

@@ -9,9 +9,12 @@ require (
 	go.opentelemetry.io/collector/confmap v1.45.0
 	go.opentelemetry.io/collector/connector v0.139.0
 	go.opentelemetry.io/collector/connector/connectortest v0.139.0
+	go.opentelemetry.io/collector/connector/xconnector v0.139.0
 	go.opentelemetry.io/collector/consumer v1.45.0
 	go.opentelemetry.io/collector/consumer/consumertest v0.139.0
+	go.opentelemetry.io/collector/consumer/xconsumer v0.139.0
 	go.opentelemetry.io/collector/pdata v1.45.0
+	go.opentelemetry.io/collector/pdata/pprofile v0.139.0
 	go.opentelemetry.io/collector/pipeline v1.45.0
 	go.uber.org/goleak v1.3.0
 )
@@ -34,11 +37,8 @@ require (
 	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
-	go.opentelemetry.io/collector/connector/xconnector v0.139.0 // indirect
-	go.opentelemetry.io/collector/consumer/xconsumer v0.139.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.45.0 // indirect
 	go.opentelemetry.io/collector/internal/fanoutconsumer v0.139.0 // indirect
-	go.opentelemetry.io/collector/pdata/pprofile v0.139.0 // indirect
 	go.opentelemetry.io/collector/pipeline/xpipeline v0.139.0 // indirect
 	go.opentelemetry.io/otel v1.38.0 // indirect
 	go.opentelemetry.io/otel/metric v1.38.0 // indirect

connector/forwardconnector/internal/metadata/generated_status.go

@@ -5,5 +5,6 @@ status:
   disable_codecov_badge: true
   class: connector
   stability:
+    development: [profiles_to_profiles]
     beta: [traces_to_traces, metrics_to_metrics, logs_to_logs]
   distributions: [core, contrib, k8s]