Add profiles support to OTLP exporter

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

exporter/otlpexporter/README.md

@@ -3,11 +3,13 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [beta]: logs   |
+| Stability     | [development]: profiles   |
+|               | [beta]: logs   |
 |               | [stable]: traces, metrics   |
 | Distributions | [core], [contrib], [k8s] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector?query=is%3Aissue%20is%3Aopen%20label%3Aexporter%2Fotlp%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector/issues?q=is%3Aopen+is%3Aissue+label%3Aexporter%2Fotlp) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector?query=is%3Aissue%20is%3Aclosed%20label%3Aexporter%2Fotlp%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector/issues?q=is%3Aclosed+is%3Aissue+label%3Aexporter%2Fotlp) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
 [beta]: https://github.com/open-telemetry/opentelemetry-collector#beta
 [stable]: https://github.com/open-telemetry/opentelemetry-collector#stable
 [core]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol

exporter/otlpexporter/cfg-schema.yaml

@@ -68,8 +68,8 @@ fields:
 - name: endpoint
   kind: string
   doc: |
-    The target to which the exporter is going to send traces or metrics,
-    using the gRPC protocol. The valid syntax is described at
+    The target to which the exporter is going to send traces, metrics, logs or
+    profiles using the gRPC protocol. The valid syntax is described at
     https://github.com/grpc/grpc/blob/master/doc/naming.md.
 - name: compression
   kind: string

exporter/otlpexporter/factory.go

@@ -15,17 +15,20 @@ import (
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterbatcher"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/exporter/exporterhelper/exporterhelperprofiles"
+	"go.opentelemetry.io/collector/exporter/exporterprofiles"
 	"go.opentelemetry.io/collector/exporter/otlpexporter/internal/metadata"
 )
 
 // NewFactory creates a factory for OTLP exporter.
 func NewFactory() exporter.Factory {
-	return exporter.NewFactory(
+	return exporterprofiles.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		exporter.WithTraces(createTraces, metadata.TracesStability),
-		exporter.WithMetrics(createMetrics, metadata.MetricsStability),
-		exporter.WithLogs(createLogs, metadata.LogsStability),
+		exporterprofiles.WithTraces(createTraces, metadata.TracesStability),
+		exporterprofiles.WithMetrics(createMetrics, metadata.MetricsStability),
+		exporterprofiles.WithLogs(createLogs, metadata.LogsStability),
+		exporterprofiles.WithProfiles(createProfilesExporter, metadata.ProfilesStability),
 	)
 }
 
@@ -104,3 +107,22 @@ func createLogs(
 		exporterhelper.WithShutdown(oce.shutdown),
 	)
 }
+
+func createProfilesExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	cfg component.Config,
+) (exporterprofiles.Profiles, error) {
+	oce := newExporter(cfg, set)
+	oCfg := cfg.(*Config)
+	return exporterhelperprofiles.NewProfilesExporter(ctx, set, cfg,
+		oce.pushProfiles,
+		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
+		exporterhelper.WithTimeout(oCfg.TimeoutConfig),
+		exporterhelper.WithRetry(oCfg.RetryConfig),
+		exporterhelper.WithQueue(oCfg.QueueConfig),
+		exporterhelper.WithBatcher(oCfg.BatcherConfig),
+		exporterhelper.WithStart(oce.start),
+		exporterhelper.WithShutdown(oce.shutdown),
+	)
+}

exporter/otlpexporter/factory_test.go

@@ -19,6 +19,7 @@ import (
 	"go.opentelemetry.io/collector/config/configretry"
 	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/exporter/exporterprofiles"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/internal/testutil"
 )
@@ -197,3 +198,143 @@ func TestCreateLogs(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, oexp)
 }
+
+func TestCreateProfiles(t *testing.T) {
+	endpoint := testutil.GetAvailableLocalAddress(t)
+	tests := []struct {
+		name            string
+		config          *Config
+		mustFailOnStart bool
+	}{
+		{
+			name: "UseSecure",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+					TLSSetting: configtls.ClientConfig{
+						Insecure: false,
+					},
+				},
+			},
+		},
+		{
+			name: "Keepalive",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+					Keepalive: &configgrpc.KeepaliveClientConfig{
+						Time:                30 * time.Second,
+						Timeout:             25 * time.Second,
+						PermitWithoutStream: true,
+					},
+				},
+			},
+		},
+		{
+			name: "NoneCompression",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint:    endpoint,
+					Compression: "none",
+				},
+			},
+		},
+		{
+			name: "GzipCompression",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint:    endpoint,
+					Compression: configcompression.TypeGzip,
+				},
+			},
+		},
+		{
+			name: "SnappyCompression",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint:    endpoint,
+					Compression: configcompression.TypeSnappy,
+				},
+			},
+		},
+		{
+			name: "ZstdCompression",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint:    endpoint,
+					Compression: configcompression.TypeZstd,
+				},
+			},
+		},
+		{
+			name: "Headers",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+					Headers: map[string]configopaque.String{
+						"hdr1": "val1",
+						"hdr2": "val2",
+					},
+				},
+			},
+		},
+		{
+			name: "NumConsumers",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+				},
+			},
+		},
+		{
+			name: "CaCert",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+					TLSSetting: configtls.ClientConfig{
+						Config: configtls.Config{
+							CAFile: filepath.Join("testdata", "test_cert.pem"),
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "CertPemFileError",
+			config: &Config{
+				ClientConfig: configgrpc.ClientConfig{
+					Endpoint: endpoint,
+					TLSSetting: configtls.ClientConfig{
+						Config: configtls.Config{
+							CAFile: "nosuchfile",
+						},
+					},
+				},
+			},
+			mustFailOnStart: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			factory := NewFactory()
+			set := exportertest.NewNopSettings()
+			consumer, err := factory.(exporterprofiles.Factory).CreateProfiles(context.Background(), set, tt.config)
+			require.NoError(t, err)
+			assert.NotNil(t, consumer)
+			err = consumer.Start(context.Background(), componenttest.NewNopHost())
+			if tt.mustFailOnStart {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			// Shutdown is called even when Start fails
+			err = consumer.Shutdown(context.Background())
+			if err != nil {
+				// Since the endpoint of OTLP exporter doesn't actually exist,
+				// exporter may already stop because it cannot connect.
+				assert.Equal(t, "rpc error: code = Canceled desc = grpc: the client connection is closing", err.Error())
+			}
+		})
+	}
+}

exporter/otlpexporter/go.mod

@@ -15,7 +15,10 @@ require (
 	go.opentelemetry.io/collector/confmap v1.17.0
 	go.opentelemetry.io/collector/consumer v0.111.0
 	go.opentelemetry.io/collector/exporter v0.111.0
+	go.opentelemetry.io/collector/exporter/exporterhelper/exporterhelperprofiles v0.0.0-00010101000000-000000000000
+	go.opentelemetry.io/collector/exporter/exporterprofiles v0.111.0
 	go.opentelemetry.io/collector/pdata v1.17.0
+	go.opentelemetry.io/collector/pdata/pprofile v0.111.0
 	go.opentelemetry.io/collector/pdata/testdata v0.111.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
@@ -49,14 +52,14 @@ require (
 	go.opentelemetry.io/collector/config/confignet v1.17.0 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.111.0 // indirect
 	go.opentelemetry.io/collector/config/internal v0.111.0 // indirect
+	go.opentelemetry.io/collector/consumer/consumererror/consumererrorprofiles v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.111.0 // indirect
 	go.opentelemetry.io/collector/consumer/consumertest v0.111.0 // indirect
-	go.opentelemetry.io/collector/exporter/exporterprofiles v0.111.0 // indirect
 	go.opentelemetry.io/collector/extension v0.111.0 // indirect
 	go.opentelemetry.io/collector/extension/auth v0.111.0 // indirect
 	go.opentelemetry.io/collector/extension/experimental/storage v0.111.0 // indirect
-	go.opentelemetry.io/collector/pdata/pprofile v0.111.0 // indirect
 	go.opentelemetry.io/collector/pipeline v0.111.0 // indirect
+	go.opentelemetry.io/collector/pipeline/pipelineprofiles v0.0.0-00010101000000-000000000000 // indirect
 	go.opentelemetry.io/collector/receiver v0.111.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.111.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 // indirect
@@ -128,6 +131,12 @@ replace go.opentelemetry.io/collector => ../..
 
 replace go.opentelemetry.io/collector/component/componentstatus => ../../component/componentstatus
 
+replace go.opentelemetry.io/collector/exporter/exporterhelper/exporterhelperprofiles => ../exporterhelper/exporterhelperprofiles
+
+replace go.opentelemetry.io/collector/consumer/consumererror/consumererrorprofiles => ../../consumer/consumererror/consumererrorprofiles
+
+replace go.opentelemetry.io/collector/pipeline/pipelineprofiles => ../../pipeline/pipelineprofiles
+
 retract (
 	v0.76.0 // Depends on retracted pdata v1.0.0-rc10 module, use v0.76.1
 	v0.69.0 // Release failed, use v0.69.1

exporter/otlpexporter/metadata.yaml

@@ -6,8 +6,9 @@ status:
   stability:
     stable: [traces, metrics]
     beta: [logs]
+    development: [profiles]
   distributions: [core, contrib, k8s]
 
 tests:
   config:
-    endpoint: otelcol:4317
+    endpoint: otelcol:4317

exporter/otlpexporter/otlp.go

@@ -25,6 +25,8 @@
 	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
+	"go.opentelemetry.io/collector/pdata/pprofile"
+	"go.opentelemetry.io/collector/pdata/pprofile/pprofileotlp"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
 )
@@ -34,12 +36,13 @@
 	config *Config
 
 	// gRPC clients and connection.
-	traceExporter  ptraceotlp.GRPCClient
-	metricExporter pmetricotlp.GRPCClient
-	logExporter    plogotlp.GRPCClient
-	clientConn     *grpc.ClientConn
-	metadata       metadata.MD
-	callOptions    []grpc.CallOption
+	traceExporter   ptraceotlp.GRPCClient
+	metricExporter  pmetricotlp.GRPCClient
+	logExporter     plogotlp.GRPCClient
+	profileExporter pprofileotlp.GRPCClient
+	clientConn      *grpc.ClientConn
+	metadata        metadata.MD
+	callOptions     []grpc.CallOption
 
 	settings component.TelemetrySettings
 
@@ -66,6 +69,7 @@
 	e.traceExporter = ptraceotlp.NewGRPCClient(e.clientConn)
 	e.metricExporter = pmetricotlp.NewGRPCClient(e.clientConn)
 	e.logExporter = plogotlp.NewGRPCClient(e.clientConn)
+	e.profileExporter = pprofileotlp.NewGRPCClient(e.clientConn)
 	headers := map[string]string{}
 	for k, v := range e.config.ClientConfig.Headers {
 		headers[k] = string(v)
@@ -133,6 +137,22 @@
 	return nil
 }
 
+func (e *baseExporter) pushProfiles(ctx context.Context, td pprofile.Profiles) error {
+	req := pprofileotlp.NewExportRequestFromProfiles(td)
+	resp, respErr := e.profileExporter.Export(e.enhanceContext(ctx), req, e.callOptions...)
+	if err := processError(respErr); err != nil {
+		return err
+	}
+	partialSuccess := resp.PartialSuccess()
+	if !(partialSuccess.ErrorMessage() == "" && partialSuccess.RejectedProfiles() == 0) {
+		e.settings.Logger.Warn("Partial success response",
+			zap.String("message", resp.PartialSuccess().ErrorMessage()),
+			zap.Int64("dropped_profiles", resp.PartialSuccess().RejectedProfiles()),
+		)
+	}
+	return nil
+}
+
 func (e *baseExporter) enhanceContext(ctx context.Context) context.Context {
 	if e.metadata.Len() > 0 {
 		return metadata.NewOutgoingContext(ctx, e.metadata)