[receiver/googlecloudpubsub] Add support for encoding extensions (#37109)

.chloggen/pubsubreceiver-encodingextensions.yaml

@@ -0,0 +1,11 @@
+change_type: enhancement
+
+component: googlecloudpubsubreceiver
+
+note: Added support for encoding extensions.
+
+issues: [37109]
+
+subtext:
+
+change_logs: [user]

receiver/googlecloudpubsubreceiver/README.md

@@ -41,29 +41,53 @@ receivers:
 
 ## Encoding
 
-You should not need to set the encoding of the subscription as the receiver will try to discover the type of the data
-by looking at the `ce-type` and `content-type` attributes of the message. Only when those attributes are not set 
-must the `encoding` field in the configuration be set. 
+The `encoding` options allows you to specify Encoding Extensions for decoding messages on the subscription. An
+extension need to be configured in the `extensions` section, and added to pipeline in the collectors configuration file.
 
-| ce-type                           | ce-datacontenttype   | encoding          | description                                    |
-|-----------------------------------|----------------------|-------------------|------------------------------------------------|
-| org.opentelemetry.otlp.traces.v1  | application/protobuf |                   | Decode OTLP trace message                      |
-| org.opentelemetry.otlp.metrics.v1 | application/protobuf |                   | Decode OTLP metric message                     |
-| org.opentelemetry.otlp.logs.v1    | application/json     |                   | Decode OTLP log message                        |
-| -                                 | -                    | otlp_proto_trace  | Decode OTLP trace message                      |
-| -                                 | -                    | otlp_proto_metric | Decode OTLP trace message                      |
-| -                                 | -                    | otlp_proto_log    | Decode OTLP trace message                      |
-| -                                 | -                    | cloud_logging     | Decode [Cloud Logging] [LogEntry] message type |
-| -                                 | -                    | raw_text          | Wrap in an OTLP log message                    |
+The following example shows how to use the text encoding extension for ingesting arbitrary text message on a 
+subscription, wrapping them in OTLP Log messages. Note that not all extensions support all signals.
+
+```yaml
+extensions:
+  text_encoding:
+    encoding: utf8
+    unmarshaling_separator: "\r?\n"
+
+service:
+  extensions: [text_encoding]
+  pipelines:
+    logs:
+      receivers: [googlecloudpubsub]
+      processors: []
+      exporters: [debug]
+```
 
-When the `encoding` configuration is set, the attributes on the message are ignored.
+The receiver also supports build in encodings for the native OTLP encodings, without the need to specify an Encoding 
+Extensions. The non OTLP build in encodings will be deprecated as soon as extensions for the formats are available.
+
+| encoding          | description                                    |
+|-------------------|------------------------------------------------|
+| otlp_proto_trace  | Decode OTLP trace message                      |
+| otlp_proto_metric | Decode OTLP trace message                      |
+| otlp_proto_log    | Decode OTLP trace message                      |
+| cloud_logging     | Decode [Cloud Logging] [LogEntry] message type |
+| raw_text          | Wrap in an OTLP log message                    |
 
 With `cloud_logging`, the receiver can be used to bring Cloud Logging messages into an OpenTelemetry pipeline. You'll
 first need to [set up a logging sink][sink-docs] with a Pub/Sub topic as its destination. Note that the `cloud_logging`
 integration is considered **alpha** as the semantic convention on some of the conversion are not stabilized yet.
 
 With `raw_text`, the receiver can be used for ingesting arbitrary text message on a Pubsub subscription, wrapping them
-in OTLP Log messages, making it a convenient way to ingest raw log lines from Pubsub.
+in OTLP Log messages.
+
+When no encoding is specified, the receiver will try to discover the type of the data by looking at the `ce-type` and
+`content-type` attributes of the message. These message attributes are set by the `googlepubsubexporter`.
+
+| ce-type                           | ce-datacontenttype   | encoding          | description                                    |
+|-----------------------------------|----------------------|-------------------|------------------------------------------------|
+| org.opentelemetry.otlp.traces.v1  | application/protobuf |                   | Decode OTLP trace message                      |
+| org.opentelemetry.otlp.metrics.v1 | application/protobuf |                   | Decode OTLP metric message                     |
+| org.opentelemetry.otlp.logs.v1    | application/protobuf |                   | Decode OTLP log message                        |
 
 [Cloud Logging]: https://cloud.google.com/logging
 [LogEntry]: https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry

receiver/googlecloudpubsubreceiver/config.go

@@ -35,51 +35,6 @@ type Config struct {
 	ClientID string `mapstructure:"client_id"`
 }
 
-func (config *Config) validateForLog() error {
-	err := config.validate()
-	if err != nil {
-		return err
-	}
-	switch config.Encoding {
-	case "":
-	case "otlp_proto_log":
-	case "raw_text":
-	case "raw_json":
-	case "cloud_logging":
-	default:
-		return fmt.Errorf("log encoding %v is not supported.  supported encoding formats include [otlp_proto_log,raw_text,raw_json,cloud_logging]", config.Encoding)
-	}
-	return nil
-}
-
-func (config *Config) validateForTrace() error {
-	err := config.validate()
-	if err != nil {
-		return err
-	}
-	switch config.Encoding {
-	case "":
-	case "otlp_proto_trace":
-	default:
-		return fmt.Errorf("trace encoding %v is not supported.  supported encoding formats include [otlp_proto_trace]", config.Encoding)
-	}
-	return nil
-}
-
-func (config *Config) validateForMetric() error {
-	err := config.validate()
-	if err != nil {
-		return err
-	}
-	switch config.Encoding {
-	case "":
-	case "otlp_proto_metric":
-	default:
-		return fmt.Errorf("metric encoding %v is not supported.  supported encoding formats include [otlp_proto_metric]", config.Encoding)
-	}
-	return nil
-}
-
 func (config *Config) validate() error {
 	if !subscriptionMatcher.MatchString(config.Subscription) {
 		return fmt.Errorf("subscription '%s' is not a valid format, use 'projects/<project_id>/subscriptions/<name>'", config.Subscription)

receiver/googlecloudpubsubreceiver/config_test.go

@@ -63,70 +63,10 @@ func TestLoadConfig(t *testing.T) {
 func TestConfigValidation(t *testing.T) {
 	factory := NewFactory()
 	c := factory.CreateDefaultConfig().(*Config)
-	assert.Error(t, c.validateForTrace())
-	assert.Error(t, c.validateForLog())
-	assert.Error(t, c.validateForMetric())
 	c.Subscription = "projects/000project/subscriptions/my-subscription"
 	assert.Error(t, c.validate())
 	c.Subscription = "projects/my-project/topics/my-topic"
 	assert.Error(t, c.validate())
 	c.Subscription = "projects/my-project/subscriptions/my-subscription"
 	assert.NoError(t, c.validate())
 }
-
-func TestTraceConfigValidation(t *testing.T) {
-	factory := NewFactory()
-	c := factory.CreateDefaultConfig().(*Config)
-	c.Subscription = "projects/my-project/subscriptions/my-subscription"
-	assert.NoError(t, c.validateForTrace())
-
-	c.Encoding = "otlp_proto_metric"
-	assert.Error(t, c.validateForTrace())
-	c.Encoding = "otlp_proto_log"
-	assert.Error(t, c.validateForTrace())
-	c.Encoding = "raw_text"
-	assert.Error(t, c.validateForTrace())
-	c.Encoding = "raw_json"
-	assert.Error(t, c.validateForTrace())
-
-	c.Encoding = "otlp_proto_trace"
-	assert.NoError(t, c.validateForTrace())
-}
-
-func TestMetricConfigValidation(t *testing.T) {
-	factory := NewFactory()
-	c := factory.CreateDefaultConfig().(*Config)
-	c.Subscription = "projects/my-project/subscriptions/my-subscription"
-	assert.NoError(t, c.validateForMetric())
-
-	c.Encoding = "otlp_proto_trace"
-	assert.Error(t, c.validateForMetric())
-	c.Encoding = "otlp_proto_log"
-	assert.Error(t, c.validateForMetric())
-	c.Encoding = "raw_text"
-	assert.Error(t, c.validateForMetric())
-	c.Encoding = "raw_json"
-	assert.Error(t, c.validateForMetric())
-
-	c.Encoding = "otlp_proto_metric"
-	assert.NoError(t, c.validateForMetric())
-}
-
-func TestLogConfigValidation(t *testing.T) {
-	factory := NewFactory()
-	c := factory.CreateDefaultConfig().(*Config)
-	c.Subscription = "projects/my-project/subscriptions/my-subscription"
-	assert.NoError(t, c.validateForLog())
-
-	c.Encoding = "otlp_proto_trace"
-	assert.Error(t, c.validateForLog())
-	c.Encoding = "otlp_proto_metric"
-	assert.Error(t, c.validateForLog())
-
-	c.Encoding = "raw_text"
-	assert.NoError(t, c.validateForLog())
-	c.Encoding = "raw_json"
-	assert.NoError(t, c.validateForLog())
-	c.Encoding = "otlp_proto_log"
-	assert.NoError(t, c.validateForLog())
-}

receiver/googlecloudpubsubreceiver/factory.go

@@ -71,7 +71,7 @@ func (factory *pubsubReceiverFactory) CreateTraces(
 	cfg component.Config,
 	consumer consumer.Traces,
 ) (receiver.Traces, error) {
-	err := cfg.(*Config).validateForTrace()
+	err := cfg.(*Config).validate()
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +89,7 @@ func (factory *pubsubReceiverFactory) CreateMetrics(
 	cfg component.Config,
 	consumer consumer.Metrics,
 ) (receiver.Metrics, error) {
-	err := cfg.(*Config).validateForMetric()
+	err := cfg.(*Config).validate()
 	if err != nil {
 		return nil, err
 	}
@@ -107,7 +107,7 @@ func (factory *pubsubReceiverFactory) CreateLogs(
 	cfg component.Config,
 	consumer consumer.Logs,
 ) (receiver.Logs, error) {
-	err := cfg.(*Config).validateForLog()
+	err := cfg.(*Config).validate()
 	if err != nil {
 		return nil, err
 	}

receiver/googlecloudpubsubreceiver/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/googleapis/gax-go/v2 v2.14.1
 	github.com/iancoleman/strcase v0.3.0
 	github.com/json-iterator/go v1.1.12
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding v0.114.0
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v0.118.1-0.20250121185328-fbefb22cc2b3
 	go.opentelemetry.io/collector/component/componenttest v0.118.1-0.20250121185328-fbefb22cc2b3
@@ -37,7 +38,7 @@ require (
 	cloud.google.com/go/iam v1.2.2 // indirect
 	cloud.google.com/go/longrunning v0.6.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -55,7 +56,7 @@ require (
 	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
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	go.einride.tech/aip v0.68.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
@@ -92,3 +93,5 @@ retract (
 	v0.76.1
 	v0.65.0
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding => ../../extension/encoding

receiver/googlecloudpubsubreceiver/internal/log_entry.go

@@ -5,7 +5,6 @@ package internal // import "github.com/open-telemetry/opentelemetry-collector-co
 
 import (
 	"bytes"
-	"context"
 	"encoding/hex"
 	stdjson "encoding/json"
 	"errors"
@@ -20,7 +19,6 @@ import (
 	jsoniter "github.com/json-iterator/go"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
-	"go.uber.org/zap"
 	"google.golang.org/genproto/googleapis/api/monitoredres"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/reflect/protoreflect"
@@ -113,7 +111,7 @@ func getLogEntryDescriptor() protoreflect.MessageDescriptor {
 // schema; this ensures that a numeric value in the input is correctly
 // translated to either an integer or a double in the output. It falls back to
 // plain JSON decoding if payload type is not available in the proto registry.
-func TranslateLogEntry(_ context.Context, _ *zap.Logger, data []byte) (pcommon.Resource, plog.LogRecord, error) {
+func TranslateLogEntry(data []byte) (pcommon.Resource, plog.LogRecord, error) {
 	lr := plog.NewLogRecord()
 	res := pcommon.NewResource()
 

receiver/googlecloudpubsubreceiver/internal/log_entry_test.go

@@ -4,7 +4,6 @@
 package internal
 
 import (
-	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -15,7 +14,6 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.uber.org/multierr"
-	"go.uber.org/zap"
 )
 
 type Log struct {
@@ -70,15 +68,12 @@ func TestTranslateLogEntry(t *testing.T) {
 	}{
 		// TODO: Add publicly shareable log test data.
 	}
-
-	logger, _ := zap.NewDevelopment()
-
 	for _, tt := range tests {
 		var errs error
 		wantRes, wantLr, err := generateLog(t, tt.want)
 		errs = multierr.Append(errs, err)
 
-		gotRes, gotLr, err := TranslateLogEntry(context.TODO(), logger, []byte(tt.input))
+		gotRes, gotLr, err := TranslateLogEntry([]byte(tt.input))
 		errs = multierr.Append(errs, err)
 		errs = multierr.Combine(errs, compareResources(wantRes, gotRes), compareLogRecords(wantLr, gotLr))