[receiver/otlpjsonfile] Add support for profiles signal (#35995)

#### Description
Add support for profiles signal to `otlpjsonfilereceiver`.

#### Link to tracking issue
Fixes #35977

#### Testing
- add `TestFileProfilesReceiver`
- update `TestFileMixedSignals`

Author: haoqixu

.chloggen/receiver-otlpjsonfile-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: otlpjsonfilereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for profiles signal
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35977]
+
+# (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: []

receiver/otlpjsonfilereceiver/README.md

@@ -3,11 +3,13 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [alpha]: traces, metrics, logs   |
+| Stability     | [development]: profiles   |
+|               | [alpha]: traces, metrics, logs   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fotlpjsonfile%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fotlpjsonfile) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fotlpjsonfile%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fotlpjsonfile) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@djaglowski](https://www.github.com/djaglowski), [@atoulme](https://www.github.com/atoulme) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
 [alpha]: https://github.com/open-telemetry/opentelemetry-collector#alpha
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->

receiver/otlpjsonfilereceiver/file.go

@@ -8,11 +8,14 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumerprofiles"
 	"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"
 	"go.opentelemetry.io/collector/receiver/receiverhelper"
+	"go.opentelemetry.io/collector/receiver/receiverprofiles"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer"
@@ -25,12 +28,13 @@ const (
 
 // NewFactory creates a factory for file receiver
 func NewFactory() receiver.Factory {
-	return receiver.NewFactory(
+	return receiverprofiles.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
-		receiver.WithLogs(createLogsReceiver, metadata.LogsStability),
-		receiver.WithTraces(createTracesReceiver, metadata.TracesStability))
+		receiverprofiles.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
+		receiverprofiles.WithLogs(createLogsReceiver, metadata.LogsStability),
+		receiverprofiles.WithTraces(createTracesReceiver, metadata.TracesStability),
+		receiverprofiles.WithProfiles(createProfilesReceiver, metadata.ProfilesStability))
 }
 
 type Config struct {
@@ -171,3 +175,24 @@ func createTracesReceiver(_ context.Context, settings receiver.Settings, configu
 
 	return &otlpjsonfilereceiver{input: input, id: settings.ID, storageID: cfg.StorageID}, nil
 }
+
+func createProfilesReceiver(_ context.Context, settings receiver.Settings, configuration component.Config, profiles consumerprofiles.Profiles) (receiverprofiles.Profiles, error) {
+	profilesUnmarshaler := &pprofile.JSONUnmarshaler{}
+	cfg := configuration.(*Config)
+	opts := make([]fileconsumer.Option, 0)
+	if cfg.ReplayFile {
+		opts = append(opts, fileconsumer.WithNoTracking())
+	}
+	input, err := cfg.Config.Build(settings.TelemetrySettings, func(ctx context.Context, token []byte, _ map[string]any) error {
+		p, _ := profilesUnmarshaler.UnmarshalProfiles(token)
+		if p.ResourceProfiles().Len() != 0 {
+			_ = profiles.ConsumeProfiles(ctx, p)
+		}
+		return nil
+	}, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &otlpjsonfilereceiver{input: input, id: settings.ID, storageID: cfg.StorageID}, nil
+}

receiver/otlpjsonfilereceiver/file_test.go

@@ -18,8 +18,10 @@ 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"
 	"go.opentelemetry.io/collector/pdata/testdata"
+	"go.opentelemetry.io/collector/receiver/receiverprofiles"
 	"go.opentelemetry.io/collector/receiver/receivertest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer"
@@ -35,6 +37,33 @@ func TestDefaultConfig(t *testing.T) {
 	require.NoError(t, componenttest.CheckConfigStruct(cfg))
 }
 
+func TestFileProfilesReceiver(t *testing.T) {
+	tempFolder := t.TempDir()
+	factory := NewFactory()
+	cfg := createDefaultConfig().(*Config)
+	cfg.Config.Include = []string{filepath.Join(tempFolder, "*")}
+	cfg.Config.StartAt = "beginning"
+	sink := new(consumertest.ProfilesSink)
+	receiver, err := factory.(receiverprofiles.Factory).CreateProfiles(context.Background(), receivertest.NewNopSettings(), cfg, sink)
+	assert.NoError(t, err)
+	err = receiver.Start(context.Background(), nil)
+	require.NoError(t, err)
+
+	pd := testdata.GenerateProfiles(5)
+	marshaler := &pprofile.JSONMarshaler{}
+	b, err := marshaler.MarshalProfiles(pd)
+	assert.NoError(t, err)
+	b = append(b, '\n')
+	err = os.WriteFile(filepath.Join(tempFolder, "profiles.json"), b, 0600)
+	assert.NoError(t, err)
+	time.Sleep(1 * time.Second)
+
+	require.Len(t, sink.AllProfiles(), 1)
+	assert.EqualValues(t, pd, sink.AllProfiles()[0])
+	err = receiver.Shutdown(context.Background())
+	assert.NoError(t, err)
+}
+
 func TestFileTracesReceiver(t *testing.T) {
 	tempFolder := t.TempDir()
 	factory := NewFactory()
@@ -213,6 +242,11 @@ func TestFileMixedSignals(t *testing.T) {
 	assert.NoError(t, err)
 	err = lr.Start(context.Background(), nil)
 	assert.NoError(t, err)
+	ps := new(consumertest.ProfilesSink)
+	pr, err := factory.(receiverprofiles.Factory).CreateProfiles(context.Background(), cs, cfg, ps)
+	assert.NoError(t, err)
+	err = pr.Start(context.Background(), nil)
+	assert.NoError(t, err)
 
 	md := testdata.GenerateMetrics(5)
 	marshaler := &pmetric.JSONMarshaler{}
@@ -226,11 +260,17 @@ func TestFileMixedSignals(t *testing.T) {
 	lmarshaler := &plog.JSONMarshaler{}
 	b3, err := lmarshaler.MarshalLogs(ld)
 	assert.NoError(t, err)
+	pd := testdata.GenerateProfiles(5)
+	pmarshaler := &pprofile.JSONMarshaler{}
+	b4, err := pmarshaler.MarshalProfiles(pd)
+	assert.NoError(t, err)
 	b = append(b, '\n')
 	b = append(b, b2...)
 	b = append(b, '\n')
 	b = append(b, b3...)
 	b = append(b, '\n')
+	b = append(b, b4...)
+	b = append(b, '\n')
 	err = os.WriteFile(filepath.Join(tempFolder, "metrics.json"), b, 0600)
 	assert.NoError(t, err)
 	time.Sleep(1 * time.Second)
@@ -241,10 +281,14 @@ func TestFileMixedSignals(t *testing.T) {
 	assert.EqualValues(t, td, ts.AllTraces()[0])
 	require.Len(t, ls.AllLogs(), 1)
 	assert.EqualValues(t, ld, ls.AllLogs()[0])
+	require.Len(t, ps.AllProfiles(), 1)
+	assert.EqualValues(t, pd, ps.AllProfiles()[0])
 	err = mr.Shutdown(context.Background())
 	assert.NoError(t, err)
 	err = tr.Shutdown(context.Background())
 	assert.NoError(t, err)
 	err = lr.Shutdown(context.Background())
 	assert.NoError(t, err)
+	err = pr.Shutdown(context.Background())
+	assert.NoError(t, err)
 }

receiver/otlpjsonfilereceiver/go.mod

@@ -12,12 +12,17 @@ require (
 	go.opentelemetry.io/collector/pdata/testdata v0.112.0
 	go.opentelemetry.io/collector/receiver v0.112.0
 	go.opentelemetry.io/collector/semconv v0.112.0 // indirect
-	go.opentelemetry.io/otel/metric v1.31.0 // indirect
-	go.opentelemetry.io/otel/trace v1.31.0 // indirect
+	go.opentelemetry.io/otel/metric v1.31.0
+	go.opentelemetry.io/otel/trace v1.31.0
 	go.uber.org/goleak v1.3.0
 )
 
-require go.opentelemetry.io/collector/consumer/consumertest v0.112.0
+require (
+	go.opentelemetry.io/collector/consumer/consumerprofiles v0.112.0
+	go.opentelemetry.io/collector/consumer/consumertest v0.112.0
+	go.opentelemetry.io/collector/pdata/pprofile v0.112.0
+	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0
+)
 
 require (
 	github.com/bmatcuk/doublestar/v4 v4.7.1 // indirect
@@ -50,13 +55,10 @@ require (
 	github.com/valyala/fastjson v1.6.4 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.112.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.112.0 // indirect
-	go.opentelemetry.io/collector/consumer/consumerprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/extension v0.112.0 // indirect
 	go.opentelemetry.io/collector/extension/experimental/storage v0.112.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.18.0 // indirect
-	go.opentelemetry.io/collector/pdata/pprofile v0.112.0 // indirect
 	go.opentelemetry.io/collector/pipeline v0.112.0 // indirect
-	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/otel v1.31.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.31.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.31.0 // indirect

receiver/otlpjsonfilereceiver/internal/metadata/generated_status.go

@@ -4,6 +4,7 @@ status:
   class: receiver
   stability:
     alpha: [traces, metrics, logs]
+    development: [profiles]
   distributions: [contrib]
   codeowners:
     active: [djaglowski, atoulme]