Add profiles support to kafka receiver (open-telemetry#41479)

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

This adds profiles support to the kafka receiver.

Related: the exporter PR:
open-telemetry#41369

Author: dmathieu

.chloggen/kafka-receiver-profiles.yaml

@@ -0,0 +1,27 @@
+# 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. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add profiles support
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [41479]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# 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: [user]

receiver/kafkareceiver/README.md

@@ -3,12 +3,14 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [beta]: metrics, logs, traces   |
+| Stability     | [development]: profiles   |
+|               | [beta]: metrics, logs, traces   |
 | Distributions | [core], [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fkafka%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fkafka) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fkafka%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fkafka) |
 | Code coverage | [![codecov](https://codecov.io/github/open-telemetry/opentelemetry-collector-contrib/graph/main/badge.svg?component=receiver_kafka)](https://app.codecov.io/gh/open-telemetry/opentelemetry-collector-contrib/tree/main/?components%5B0%5D=receiver_kafka&displayType=list) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@pavolloffay](https://www.github.com/pavolloffay), [@MovieStoreGuy](https://www.github.com/MovieStoreGuy), [@axw](https://www.github.com/axw) |
 
+[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
@@ -48,6 +50,9 @@ The following settings can be optionally configured:
 - `traces`
   - `topic` (default = otlp\_spans): The name of the Kafka topic from which to consume traces.
   - `encoding` (default = otlp\_proto): The encoding for the Kafka topic. See [Supported encodings](#supported-encodings).
+- `profiles`
+  - `topic` (default = otlp\_profiles): The name of the Kafka topic from which to consume profiles.
+  - `encoding` (default = otlp\_proto): The encoding for the Kafka topic. See [Supported encodings](#supported-encodings).
 - `topic` (Deprecated [v0.124.0]: use `logs::topic`, `traces::topic`, or `metrics::topic`).
    If this is set, it will take precedence over the default value for those fields.
 - `encoding` (Deprecated [v0.124.0]: use `logs::encoding`, `traces::encoding`, or `metrics::encoding`).
@@ -109,10 +114,10 @@ The following settings can be optionally configured:
     **Note: this can block the entire partition in case a message processing returns a permanent error**
 - `header_extraction`:
   - `extract_headers` (default = false): Allows user to attach header fields to resource attributes in otel pipeline
-  - `headers` (default = []): List of headers they'd like to extract from kafka record. 
-  **Note: Matching pattern will be `exact`. Regexes are not supported as of now.** 
+  - `headers` (default = []): List of headers they'd like to extract from kafka record.
+  **Note: Matching pattern will be `exact`. Regexes are not supported as of now.**
 - `error_backoff`: [BackOff](https://github.com/open-telemetry/opentelemetry-collector/blob/v0.116.0/config/configretry/backoff.go#L27-L43) configuration in case of errors
-  - `enabled`: (default = false) Whether to enable backoff when next consumers return errors 
+  - `enabled`: (default = false) Whether to enable backoff when next consumers return errors
   - `initial_interval`: The time to wait after the first error before retrying
   - `max_interval`: The upper bound on backoff interval between consecutive retries
   - `multiplier`: The value multiplied by the backoff interval bounds
@@ -190,7 +195,7 @@ be configured to extract and attach specific headers as resource attributes. e.g
 ```yaml
 receivers:
   kafka:
-    header_extraction: 
+    header_extraction:
       extract_headers: true
       headers: ["header1", "header2"]
 ```

receiver/kafkareceiver/config.go

@@ -27,6 +27,9 @@ type Config struct {
 	// Traces holds configuration about how traces should be consumed.
 	Traces TopicEncodingConfig `mapstructure:"traces"`
 
+	// Profiles holds configuration about how profiles should be consumed.
+	Profiles TopicEncodingConfig `mapstructure:"profiles"`
+
 	// Topic holds the name of the Kafka topic from which to consume data.
 	//
 	// Topic has no default. If explicitly specified, it will take precedence
@@ -80,6 +83,9 @@ func (c *Config) Unmarshal(conf *confmap.Conf) error {
 		if zeroConfig.Traces.Topic == "" {
 			c.Traces.Topic = c.Topic
 		}
+		if zeroConfig.Profiles.Topic == "" {
+			c.Profiles.Topic = c.Topic
+		}
 	}
 	if c.Encoding != "" {
 		if zeroConfig.Logs.Encoding == "" {
@@ -91,6 +97,9 @@ func (c *Config) Unmarshal(conf *confmap.Conf) error {
 		if zeroConfig.Traces.Encoding == "" {
 			c.Traces.Encoding = c.Encoding
 		}
+		if zeroConfig.Profiles.Encoding == "" {
+			c.Profiles.Encoding = c.Encoding
+		}
 	}
 
 	// Set OnPermanentError default value to inherit from OnError for backward compatibility
@@ -119,6 +128,7 @@ type TopicEncodingConfig struct {
 	//  - "otlp_spans" for traces
 	//  - "otlp_metrics" for metrics
 	//  - "otlp_logs" for logs
+	//  - "otlp_profiles" for profiles
 	Topic string `mapstructure:"topic"`
 
 	// Encoding holds the expected encoding of messages for the signal type

receiver/kafkareceiver/config_test.go

@@ -58,6 +58,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "spans",
+					Encoding: "otlp_proto",
+				},
 				Topic: "spans",
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled: false,
@@ -88,6 +92,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "legacy_topic",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "legacy_topic",
+					Encoding: "otlp_proto",
+				},
 				Topic: "legacy_topic",
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled: false,
@@ -111,6 +119,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "legacy_encoding",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "legacy_encoding",
+				},
 				Encoding: "legacy_encoding",
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled: false,
@@ -158,6 +170,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "otlp_proto",
+				},
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled:         true,
 					InitialInterval: 1 * time.Second,
@@ -189,6 +205,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "otlp_proto",
+				},
 				ErrorBackOff: configretry.BackOffConfig{
 					Enabled: false,
 				},
@@ -211,6 +231,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "otlp_proto",
+				},
 				MessageMarking: MessageMarking{
 					After:            true,
 					OnError:          true,
@@ -238,6 +262,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "otlp_proto",
+				},
 				MessageMarking: MessageMarking{
 					After:            false,
 					OnError:          false,
@@ -265,6 +293,10 @@ func TestLoadConfig(t *testing.T) {
 					Topic:    "otlp_spans",
 					Encoding: "otlp_proto",
 				},
+				Profiles: TopicEncodingConfig{
+					Topic:    "otlp_profiles",
+					Encoding: "otlp_proto",
+				},
 				MessageMarking: MessageMarking{
 					After:            true,
 					OnError:          true,

receiver/kafkareceiver/documentation.md

@@ -278,6 +278,21 @@ Number of metric points failed to be unmarshaled
 | topic | The Kafka topic. | Any Str |
 | partition | The Kafka topic partition. | Any Int |
 
+### otelcol_kafka_receiver_unmarshal_failed_profiles
+
+Number of profiles failed to be unmarshaled
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| 1 | Sum | Int | true |
+
+#### Attributes
+
+| Name | Description | Values |
+| ---- | ----------- | ------ |
+| topic | The Kafka topic. | Any Str |
+| partition | The Kafka topic partition. | Any Int |
+
 ### otelcol_kafka_receiver_unmarshal_failed_spans
 
 Number of spans failed to be unmarshaled

receiver/kafkareceiver/encoding.go

@@ -11,6 +11,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"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/receiver"
 
@@ -127,6 +128,24 @@ func loadEncodingExtension[T any](host component.Host, encoding, signalType stri
 	return unmarshaler, nil
 }
 
+func newProfilesUnmarshaler(encoding string, _ receiver.Settings, host component.Host) (pprofile.Unmarshaler, error) {
+	// Extensions take precedence.
+	if unmarshaler, err := loadEncodingExtension[pprofile.Unmarshaler](host, encoding, "profiles"); err != nil {
+		if !errors.Is(err, errInvalidComponentType) && !errors.Is(err, errUnknownEncodingExtension) {
+			return nil, err
+		}
+	} else {
+		return unmarshaler, nil
+	}
+	switch encoding {
+	case "otlp_proto":
+		return &pprofile.ProtoUnmarshaler{}, nil
+	case "otlp_json":
+		return &pprofile.JSONUnmarshaler{}, nil
+	}
+	return nil, fmt.Errorf("unrecognized profiles encoding %q", encoding)
+}
+
 // encodingToComponentID attempts to parse the encoding string as a component ID.
 func encodingToComponentID(encoding string) (*component.ID, error) {
 	var id component.ID

receiver/kafkareceiver/encoding_test.go

@@ -16,12 +16,15 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"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"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"golang.org/x/text/encoding/unicode"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pprofiletest"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/ptracetest"
 	zipkinthriftconverter "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin/zipkinthriftconverter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/zipkin/zipkinv2"
@@ -41,6 +44,10 @@ var (
 		component.Component
 		ptrace.Unmarshaler
 	}
+	customProfilesUnmarshalerExtension struct {
+		component.Component
+		pprofile.Unmarshaler
+	}
 )
 
 func TestNewLogsUnmarshaler(t *testing.T) {
@@ -376,6 +383,60 @@ func TestNewTracesUnmarshalerExtension(t *testing.T) {
 	assert.Nil(t, u)
 }
 
+func TestNewProfilesUnmarshaler(t *testing.T) {
+	profiles := testdata.GenerateProfiles(3)
+
+	otlpProtoProfiles, err := (&pprofile.ProtoMarshaler{}).MarshalProfiles(profiles)
+	require.NoError(t, err)
+	otlpJSONProfiles, err := (&pprofile.JSONMarshaler{}).MarshalProfiles(profiles)
+	require.NoError(t, err)
+
+	for _, tc := range []struct {
+		encoding string
+		input    []byte
+		check    func(*testing.T, pprofile.Profiles)
+	}{
+		{
+			encoding: "otlp_proto",
+			input:    otlpProtoProfiles,
+			check: func(t *testing.T, actual pprofile.Profiles) {
+				assert.NoError(t, pprofiletest.CompareProfiles(profiles, actual))
+			},
+		},
+		{
+			encoding: "otlp_json",
+			input:    otlpJSONProfiles,
+			check: func(t *testing.T, actual pprofile.Profiles) {
+				assert.NoError(t, pprofiletest.CompareProfiles(profiles, actual))
+			},
+		},
+	} {
+		t.Run(tc.encoding, func(t *testing.T) {
+			u := mustNewProfilesUnmarshaler(t, tc.encoding, componenttest.NewNopHost())
+			out, err := u.UnmarshalProfiles(tc.input)
+			require.NoError(t, err)
+			tc.check(t, out)
+		})
+	}
+}
+
+func TestNewProfilesUnmarshalerExtension(t *testing.T) {
+	settings := receivertest.NewNopSettings(metadata.Type)
+
+	// Verify extensions take precedence over built-in unmarshalers.
+	u := mustNewProfilesUnmarshaler(t, "otlp_proto", extensionsHost{
+		component.MustNewID("otlp_proto"): &customProfilesUnmarshalerExtension,
+	})
+	assert.Equal(t, &customProfilesUnmarshalerExtension, u)
+
+	// Specifying an extension for a different type should fail fast.
+	u, err := newProfilesUnmarshaler("not_profiles", settings, extensionsHost{
+		component.MustNewID("not_profiles"): &customLogsUnmarshalerExtension,
+	})
+	require.EqualError(t, err, `extension "not_profiles" is not a profiles unmarshaler`)
+	assert.Nil(t, u)
+}
+
 func mustNewLogsUnmarshaler(tb testing.TB, encoding string, host component.Host) plog.Unmarshaler {
 	settings := receivertest.NewNopSettings(metadata.Type)
 	u, err := newLogsUnmarshaler(encoding, settings, host)
@@ -397,6 +458,13 @@ func mustNewTracesUnmarshaler(tb testing.TB, encoding string, host component.Hos
 	return u
 }
 
+func mustNewProfilesUnmarshaler(tb testing.TB, encoding string, host component.Host) pprofile.Unmarshaler {
+	settings := receivertest.NewNopSettings(metadata.Type)
+	u, err := newProfilesUnmarshaler(encoding, settings, host)
+	require.NoError(tb, err)
+	return u
+}
+
 type extensionsHost map[component.ID]component.Component
 
 func (h extensionsHost) GetExtensions() map[component.ID]component.Component {