POC: Add support for profiles within experimental packages

componentexperimental/Makefile

@@ -0,0 +1 @@
+include ../Makefile.Common

componentexperimental/config.go

@@ -0,0 +1,16 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package componentexperimental // import "go.opentelemetry.io/collector/componentexperimental"
+
+import "go.opentelemetry.io/collector/component"
+
+func mustNewDataType(strType string) component.DataType {
+	return component.MustNewType(strType)
+}
+
+// Currently supported data types. Add new data types here when new types are supported in the future.
+var (
+	// DataTypeProfiles is the data type tag for profiles.
+	DataTypeProfiles = mustNewDataType("profiles")
+)

componentexperimental/go.mod

@@ -0,0 +1,29 @@
+module go.opentelemetry.io/collector/componentexperimental
+
+go 1.21.0
+
+require go.opentelemetry.io/collector/component v0.102.0
+
+require (
+	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.102.0 // indirect
+	go.opentelemetry.io/collector/confmap v0.102.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.9.0 // indirect
+	go.opentelemetry.io/otel v1.27.0 // indirect
+	go.opentelemetry.io/otel/metric v1.27.0 // indirect
+	go.opentelemetry.io/otel/trace v1.27.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/sys v0.20.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/grpc v1.64.0 // indirect
+	google.golang.org/protobuf v1.34.1 // indirect
+)

connectorexperimental/Makefile

@@ -0,0 +1 @@
+include ../Makefile.Common

connectorexperimental/fanoutconsumerexperimental/profiles.go

@@ -0,0 +1,82 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package fanoutconsumerexperimental // import "go.opentelemetry.io/collector/connectorexperimental/fanoutconsumerexperimental"
+
+import (
+	"context"
+
+	"go.uber.org/multierr"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumerexperimental"
+	"go.opentelemetry.io/collector/pdata/pprofile"
+)
+
+// NewProfiles wraps multiple profile consumers in a single one.
+// It fanouts the incoming data to all the consumers, and does smart routing:
+//   - Clones only to the consumer that needs to mutate the data.
+//   - If all consumers needs to mutate the data one will get the original mutable data.
+func NewProfiles(tcs []consumerexperimental.Profiles) consumerexperimental.Profiles {
+	// Don't wrap if there is only one non-mutating consumer.
+	if len(tcs) == 1 && !tcs[0].Capabilities().MutatesData {
+		return tcs[0]
+	}
+
+	tc := &profilesConsumer{}
+	for i := 0; i < len(tcs); i++ {
+		if tcs[i].Capabilities().MutatesData {
+			tc.mutable = append(tc.mutable, tcs[i])
+		} else {
+			tc.readonly = append(tc.readonly, tcs[i])
+		}
+	}
+	return tc
+}
+
+type profilesConsumer struct {
+	mutable  []consumerexperimental.Profiles
+	readonly []consumerexperimental.Profiles
+}
+
+func (tsc *profilesConsumer) Capabilities() consumer.Capabilities {
+	// If all consumers are mutating, then the original data will be passed to one of them.
+	return consumer.Capabilities{MutatesData: len(tsc.mutable) > 0 && len(tsc.readonly) == 0}
+}
+
+// ConsumeProfiles exports the pprofile.Profiles to all consumers wrapped by the current one.
+func (tsc *profilesConsumer) ConsumeProfiles(ctx context.Context, td pprofile.Profiles) error {
+	var errs error
+
+	if len(tsc.mutable) > 0 {
+		// Clone the data before sending to all mutating consumers except the last one.
+		for i := 0; i < len(tsc.mutable)-1; i++ {
+			errs = multierr.Append(errs, tsc.mutable[i].ConsumeProfiles(ctx, cloneProfiles(td)))
+		}
+		// Send data as is to the last mutating consumer only if there are no other non-mutating consumers and the
+		// data is mutable. Never share the same data between a mutating and a non-mutating consumer since the
+		// non-mutating consumer may process data async and the mutating consumer may change the data before that.
+		lastConsumer := tsc.mutable[len(tsc.mutable)-1]
+		if len(tsc.readonly) == 0 && !td.IsReadOnly() {
+			errs = multierr.Append(errs, lastConsumer.ConsumeProfiles(ctx, td))
+		} else {
+			errs = multierr.Append(errs, lastConsumer.ConsumeProfiles(ctx, cloneProfiles(td)))
+		}
+	}
+
+	// Mark the data as read-only if it will be sent to more than one read-only consumer.
+	if len(tsc.readonly) > 1 && !td.IsReadOnly() {
+		td.MarkReadOnly()
+	}
+	for _, tc := range tsc.readonly {
+		errs = multierr.Append(errs, tc.ConsumeProfiles(ctx, td))
+	}
+
+	return errs
+}
+
+func cloneProfiles(td pprofile.Profiles) pprofile.Profiles {
+	clonedProfiles := pprofile.NewProfiles()
+	td.CopyTo(clonedProfiles)
+	return clonedProfiles
+}

connectorexperimental/go.mod

@@ -0,0 +1,48 @@
+module go.opentelemetry.io/collector/connectorexperimental
+
+go 1.21.0
+
+require (
+	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/component v0.102.0
+	go.opentelemetry.io/collector/consumer v0.102.0
+	go.opentelemetry.io/collector/consumerexperimental v0.0.0-00010101000000-000000000000
+	go.opentelemetry.io/collector/pdata/pprofile v0.0.0-20240529223953-eaab76e46d38
+	go.opentelemetry.io/collector/pdata/testdata v0.102.0
+	go.uber.org/multierr v1.11.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	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/config/configtelemetry v0.102.0 // indirect
+	go.opentelemetry.io/collector/confmap v0.102.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.9.0 // indirect
+	go.opentelemetry.io/otel v1.27.0 // indirect
+	go.opentelemetry.io/otel/metric v1.27.0 // indirect
+	go.opentelemetry.io/otel/trace v1.27.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/net v0.25.0 // indirect
+	golang.org/x/sys v0.20.0 // indirect
+	golang.org/x/text v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240520151616-dc85e6b867a5 // indirect
+	google.golang.org/grpc v1.64.0 // indirect
+	google.golang.org/protobuf v1.34.1 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+replace go.opentelemetry.io/collector/pdata => ../pdata
+
+replace go.opentelemetry.io/collector/pdata/pprofile => ../pdata/pprofile
+
+replace go.opentelemetry.io/collector/consumerexperimental => ../consumerexperimental