Profiles support in OTLP receiver

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

cmd/mdatagen/go.mod

@@ -101,3 +101,5 @@ replace go.opentelemetry.io/collector/consumer/consumertest => ../../consumer/co
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

cmd/mdatagen/validate.go

@@ -95,15 +95,23 @@ func (s *Status) validateStability() error {
 			if c != "metrics" &&
 				c != "traces" &&
 				c != "logs" &&
+				c != "profiles" &&
 				c != "traces_to_traces" &&
 				c != "traces_to_metrics" &&
 				c != "traces_to_logs" &&
+				c != "traces_to_profiles" &&
 				c != "metrics_to_traces" &&
 				c != "metrics_to_metrics" &&
 				c != "metrics_to_logs" &&
+				c != "metrics_to_profiles" &&
 				c != "logs_to_traces" &&
 				c != "logs_to_metrics" &&
 				c != "logs_to_logs" &&
+				c != "logs_to_profiles" &&
+				c != "profiles_to_profiles" &&
+				c != "profiles_to_traces" &&
+				c != "profiles_to_metrics" &&
+				c != "profiles_to_logs" &&
 				c != "extension" {
 				errs = errors.Join(errs, fmt.Errorf("invalid component: %v", c))
 			}

component/componenttest/obsreporttest.go

@@ -116,6 +116,12 @@
 	return tts.prometheusChecker.checkReceiverMetrics(tts.id, protocol, acceptedMetricPoints, droppedMetricPoints)
 }
 
+// CheckReceiverProfiles checks that for the current exported values for profiles receiver metrics match given values.
+// Note: SetupTelemetry must be called before this function.
+func (tts *TestTelemetry) CheckReceiverProfiles(protocol string, acceptedSamples, droppedSamples int64) error {
+	return tts.prometheusChecker.checkReceiverProfiles(tts.id, protocol, acceptedSamples, droppedSamples)
+}
+
 // CheckScraperMetrics checks that for the current exported values for metrics scraper metrics match given values.
 // Note: SetupTelemetry must be called before this function.
 func (tts *TestTelemetry) CheckScraperMetrics(receiver component.ID, scraper component.ID, scrapedMetricPoints, erroredMetricPoints int64) error {

component/componenttest/otelprometheuschecker.go

@@ -41,6 +41,10 @@ func (pc *prometheusChecker) checkReceiverMetrics(receiver component.ID, protoco
 	return pc.checkReceiver(receiver, "metric_points", protocol, accepted, dropped)
 }
 
+func (pc *prometheusChecker) checkReceiverProfiles(receiver component.ID, protocol string, accepted, dropped int64) error {
+	return pc.checkReceiver(receiver, "samples", protocol, accepted, dropped)
+}
+
 func (pc *prometheusChecker) checkReceiver(receiver component.ID, datatype, protocol string, acceptedMetricPoints, droppedMetricPoints int64) error {
 	receiverAttrs := attributesForReceiverMetrics(receiver, protocol)
 	return multierr.Combine(

component/componenttest/otelprometheuschecker_test.go

@@ -87,6 +87,11 @@ func TestPromChecker(t *testing.T) {
 		"metrics from Receiver Logs should be valid",
 	)
 
+	assert.NoError(t,
+		pc.checkReceiverProfiles(receiver, transport, 42, 13),
+		"metrics from Receiver Profiles should be valid",
+	)
+
 	assert.NoError(t,
 		pc.checkProcessorTraces(processor, 42, 13, 7, 5),
 		"metrics from Receiver Traces should be valid",

component/componenttest/testdata/prometheus_response

@@ -76,6 +76,12 @@ otelcol_scraper_scraped_metric_points{receiver="fakeReceiver",scraper="fakeScrap
 # HELP otelcol_scraper_errored_metric_points Number of metric points that were unable to be scraped.
 # TYPE otelcol_scraper_errored_metric_points counter
 otelcol_scraper_errored_metric_points{receiver="fakeReceiver",scraper="fakeScraper"} 41
+# HELP otelcol_receiver_accepted_samples Number of profiles samples successfully pushed into the pipeline.
+# TYPE otelcol_receiver_accepted_samples counter
+otelcol_receiver_accepted_samples{receiver="fakeReceiver",transport="fakeTransport"} 42
+# HELP otelcol_receiver_refused_samples Number of profiles samples that could not be pushed into the pipeline.
+# TYPE otelcol_receiver_refused_samples counter
+otelcol_receiver_refused_samples{receiver="fakeReceiver",transport="fakeTransport"} 13
 # HELP gauge_metric A simple gauge metric
 # TYPE gauge_metric gauge
 gauge_metric 49

exporter/debugexporter/go.mod

@@ -103,3 +103,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../../compone
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ../exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

exporter/exporterprofiles/go.mod

@@ -70,3 +70,5 @@ replace go.opentelemetry.io/collector/exporter => ../
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

exporter/go.mod

@@ -102,3 +102,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../component/
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ./exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../component/componentprofiles

exporter/loggingexporter/go.mod

@@ -108,3 +108,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../../compone
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ../exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

exporter/nopexporter/go.mod

@@ -97,3 +97,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../../compone
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ../exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

exporter/otlpexporter/go.mod

@@ -156,3 +156,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../../compone
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ../exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

exporter/otlphttpexporter/go.mod

@@ -153,3 +153,5 @@ replace go.opentelemetry.io/collector/component/componentstatus => ../../compone
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../../receiver/receiverprofiles
 
 replace go.opentelemetry.io/collector/exporter/exporterprofiles => ../exporterprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles

internal/obsreportconfig/obsmetrics/obs_receiver.go

@@ -27,6 +27,11 @@ const (
 	// RefusedLogRecordsKey used to identify log records refused (ie.: not ingested) by the
 	// Collector.
 	RefusedLogRecordsKey = "refused_log_records"
+
+	// AcceptedSamplesKey used to identify profiling samples accepted by the Collector.
+	AcceptedSamplesKey = "accepted_spans"
+	// RefusedSpansKey used to identify spans refused (ie.: not ingested) by the Collector.
+	RefusedSamplesKey = "refused_spans"
 )
 
 var (

pdata/pprofile/pb.go

@@ -0,0 +1,31 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package pprofile // import "go.opentelemetry.io/collector/pdata/pprofile"
+
+import (
+	"go.opentelemetry.io/collector/pdata/internal"
+	otlpprofile "go.opentelemetry.io/collector/pdata/internal/data/protogen/profiles/v1experimental"
+)
+
+var _ MarshalSizer = (*ProtoMarshaler)(nil)
+
+type ProtoMarshaler struct{}
+
+func (e *ProtoMarshaler) MarshalProfiles(td Profiles) ([]byte, error) {
+	pb := internal.ProfilesToProto(internal.Profiles(td))
+	return pb.Marshal()
+}
+
+func (e *ProtoMarshaler) ProfilesSize(td Profiles) int {
+	pb := internal.ProfilesToProto(internal.Profiles(td))
+	return pb.Size()
+}
+
+type ProtoUnmarshaler struct{}
+
+func (d *ProtoUnmarshaler) UnmarshalProfiles(buf []byte) (Profiles, error) {
+	pb := otlpprofile.ProfilesData{}
+	err := pb.Unmarshal(buf)
+	return Profiles(internal.ProfilesFromProto(pb)), err
+}

pdata/pprofile/pb_test.go

@@ -0,0 +1,78 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package pprofile
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProtoProfilesUnmarshalerError(t *testing.T) {
+	p := &ProtoUnmarshaler{}
+	_, err := p.UnmarshalProfiles([]byte("+$%"))
+	assert.Error(t, err)
+}
+
+func TestProtoSizer(t *testing.T) {
+	marshaler := &ProtoMarshaler{}
+	td := NewProfiles()
+	td.ResourceProfiles().AppendEmpty().
+		ScopeProfiles().AppendEmpty().
+		Profiles().AppendEmpty().
+		Profile().
+		StringTable().Append("foobar")
+
+	size := marshaler.ProfilesSize(td)
+
+	bytes, err := marshaler.MarshalProfiles(td)
+	require.NoError(t, err)
+	assert.Equal(t, len(bytes), size)
+}
+
+func TestProtoSizerEmptyProfiles(t *testing.T) {
+	sizer := &ProtoMarshaler{}
+	assert.Equal(t, 0, sizer.ProfilesSize(NewProfiles()))
+}
+
+func BenchmarkProfilesToProto(b *testing.B) {
+	marshaler := &ProtoMarshaler{}
+	profiles := generateBenchmarkProfiles(128)
+	b.ResetTimer()
+	for n := 0; n < b.N; n++ {
+		buf, err := marshaler.MarshalProfiles(profiles)
+		require.NoError(b, err)
+		assert.NotEmpty(b, buf)
+	}
+}
+
+func BenchmarkProfilesFromProto(b *testing.B) {
+	marshaler := &ProtoMarshaler{}
+	unmarshaler := &ProtoUnmarshaler{}
+	baseProfiles := generateBenchmarkProfiles(128)
+	buf, err := marshaler.MarshalProfiles(baseProfiles)
+	require.NoError(b, err)
+	assert.NotEmpty(b, buf)
+	b.ResetTimer()
+	b.ReportAllocs()
+	for n := 0; n < b.N; n++ {
+		profiles, err := unmarshaler.UnmarshalProfiles(buf)
+		require.NoError(b, err)
+		assert.Equal(b, baseProfiles.ResourceProfiles().Len(), profiles.ResourceProfiles().Len())
+	}
+}
+
+func generateBenchmarkProfiles(samplesCount int) Profiles {
+	md := NewProfiles()
+	ilm := md.ResourceProfiles().AppendEmpty().ScopeProfiles().AppendEmpty().Profiles().AppendEmpty().Profile()
+	ilm.Sample().EnsureCapacity(samplesCount)
+	for i := 0; i < samplesCount; i++ {
+		im := ilm.Sample().AppendEmpty()
+		im.LocationIndex().Append(1)
+		im.SetLocationsStartIndex(2)
+		im.SetLocationsLength(10)
+	}
+	return md
+}

pdata/pprofile/profiles_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"go.opentelemetry.io/collector/pdata/internal/data"
+	otlpcollectorprofile "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/profiles/v1experimental"
+	otlpprofile "go.opentelemetry.io/collector/pdata/internal/data/protogen/profiles/v1experimental"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
@@ -23,6 +25,65 @@ func TestReadOnlyProfilesInvalidUsage(t *testing.T) {
 	assert.Panics(t, func() { res.Attributes().PutStr("k2", "v2") })
 }
 
+func TestSampleCount(t *testing.T) {
+	profiles := NewProfiles()
+	assert.EqualValues(t, 0, profiles.SampleCount())
+
+	rs := profiles.ResourceProfiles().AppendEmpty()
+	assert.EqualValues(t, 0, profiles.SampleCount())
+
+	ils := rs.ScopeProfiles().AppendEmpty()
+	assert.EqualValues(t, 0, profiles.SampleCount())
+
+	ps := ils.Profiles().AppendEmpty().Profile()
+	assert.EqualValues(t, 0, profiles.SampleCount())
+
+	ps.Sample().AppendEmpty()
+	assert.EqualValues(t, 1, profiles.SampleCount())
+
+	rms := profiles.ResourceProfiles()
+	rms.EnsureCapacity(3)
+	rms.AppendEmpty().ScopeProfiles().AppendEmpty()
+	ilss := rms.AppendEmpty().ScopeProfiles().AppendEmpty().Profiles().AppendEmpty().Profile().Sample()
+	for i := 0; i < 5; i++ {
+		ilss.AppendEmpty()
+	}
+	// 5 + 1 (from rms.At(0) initialized first)
+	assert.EqualValues(t, 6, profiles.SampleCount())
+}
+
+func TestSampleCountWithEmpty(t *testing.T) {
+	assert.EqualValues(t, 0, newProfiles(&otlpcollectorprofile.ExportProfilesServiceRequest{
+		ResourceProfiles: []*otlpprofile.ResourceProfiles{{}},
+	}).SampleCount())
+	assert.EqualValues(t, 0, newProfiles(&otlpcollectorprofile.ExportProfilesServiceRequest{
+		ResourceProfiles: []*otlpprofile.ResourceProfiles{
+			{
+				ScopeProfiles: []*otlpprofile.ScopeProfiles{{}},
+			},
+		},
+	}).SampleCount())
+	assert.EqualValues(t, 1, newProfiles(&otlpcollectorprofile.ExportProfilesServiceRequest{
+		ResourceProfiles: []*otlpprofile.ResourceProfiles{
+			{
+				ScopeProfiles: []*otlpprofile.ScopeProfiles{
+					{
+						Profiles: []*otlpprofile.ProfileContainer{
+							{
+								Profile: otlpprofile.Profile{
+									Sample: []otlpprofile.Sample{
+										{},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}).SampleCount())
+}
+
 func BenchmarkProfilesUsage(b *testing.B) {
 	profiles := NewProfiles()
 	fillTestResourceProfilesSlice(profiles.ResourceProfiles())

pdata/testdata/profile.go

@@ -37,10 +37,28 @@ func fillProfileOne(profile pprofile.ProfileContainer) {
 	profile.SetStartTime(profileStartTimestamp)
 	profile.SetEndTime(profileEndTimestamp)
 	profile.SetDroppedAttributesCount(1)
+
+	sample := profile.Profile().Sample().AppendEmpty()
+	sample.LocationIndex().Append(1)
+	sample.SetLocationsStartIndex(2)
+	sample.SetLocationsLength(10)
+	sample.SetStacktraceIdIndex(3)
+	sample.Value().Append(4)
+	sample.SetLink(42)
+	sample.Attributes().Append(5)
 }
 
 func fillProfileTwo(profile pprofile.ProfileContainer) {
 	profile.SetProfileID([16]byte{0x02, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10})
 	profile.SetStartTime(profileStartTimestamp)
 	profile.SetEndTime(profileEndTimestamp)
+
+	sample := profile.Profile().Sample().AppendEmpty()
+	sample.LocationIndex().Append(6)
+	sample.SetLocationsStartIndex(7)
+	sample.SetLocationsLength(20)
+	sample.SetStacktraceIdIndex(8)
+	sample.Value().Append(9)
+	sample.SetLink(44)
+	sample.Attributes().Append(10)
 }

receiver/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector v0.109.0
 	go.opentelemetry.io/collector/component v0.109.0
+	go.opentelemetry.io/collector/component/componentprofiles v0.108.1
 	go.opentelemetry.io/collector/config/configtelemetry v0.109.0
 	go.opentelemetry.io/collector/consumer v0.109.0
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.109.0
@@ -78,3 +79,5 @@ replace go.opentelemetry.io/collector/consumer/consumertest => ../consumer/consu
 replace go.opentelemetry.io/collector/component/componentstatus => ../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ./receiverprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../component/componentprofiles

receiver/nopreceiver/go.mod

@@ -85,3 +85,5 @@ replace go.opentelemetry.io/collector/consumer/consumertest => ../../consumer/co
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
 
 replace go.opentelemetry.io/collector/receiver/receiverprofiles => ../receiverprofiles
+
+replace go.opentelemetry.io/collector/component/componentprofiles => ../../component/componentprofiles