[receiver/googlecloudpubsub] Add support for encoding extensions (#37109) #37137
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
alexvanboxel
force-pushed
the
pubsubreceiver-encodingextensions
branch
2 times, most recently
from
8952521 to
fa9e362
Compare
…n-telemetry#37109) Added support for encoding extensions. Setting the encoding field in the config now references the extension. If it didn't find the extension it will fall back to searching the internal encoders. To make the build in encoders consistent with the extensions they now have the same interface. The README is adapted accordingly.
alexvanboxel
force-pushed
the
pubsubreceiver-encodingextensions
branch
from
fa9e362 to
29ffa42
Compare
This was referenced Jan 10, 2025
atoulme
approved these changes
Jan 10, 2025
I am the codeowner (sorry, should have stated that in the PR) |
spiderman meme here |
atoulme
added
the
ready to merge
Comment on lines
113
to
184
| if receiver.config.Encoding != "" { | ||
| if receiver.consumerCount() > 1 { | ||
| return errors.New("cannot start receiver: multiple consumers were attached, but encoding was specified") | ||
| } | ||
| encodingID := convertEncoding(receiver.config.Encoding) | ||
| if encodingID == unknown { | ||
| extensionID := component.ID{} | ||
| err := extensionID.UnmarshalText([]byte(receiver.config.Encoding)) | ||
| if err != nil { | ||
| return errors.New("cannot start receiver: neither a build in encoder, or an extension") | ||
| } | ||
| extensions := host.GetExtensions() | ||
| if extension, ok := extensions[extensionID]; ok { | ||
| if receiver.tracesConsumer != nil { | ||
| receiver.tracesUnmarshaler, ok = extension.(encoding.TracesUnmarshalerExtension) | ||
| if !ok { | ||
| return fmt.Errorf("cannot start receiver: extension %q is not a trace unmarshaler", extensionID) | ||
| } | ||
| } | ||
| if receiver.logsConsumer != nil { | ||
| receiver.logsUnmarshaler, ok = extension.(encoding.LogsUnmarshalerExtension) | ||
| if !ok { | ||
| return fmt.Errorf("cannot start receiver: extension %q is not a logs unmarshaler", extensionID) | ||
| } | ||
| } | ||
| if receiver.metricsConsumer != nil { | ||
| receiver.metricsUnmarshaler, ok = extension.(encoding.MetricsUnmarshalerExtension) | ||
| if !ok { | ||
| return fmt.Errorf("cannot start receiver: extension %q is not a metrics unmarshaler", extensionID) | ||
| } | ||
| } | ||
| } else { | ||
| return fmt.Errorf("cannot start receiver: extension %q not found", extensionID) | ||
| } | ||
| } else { | ||
| if receiver.tracesConsumer != nil { | ||
| switch encodingID { | ||
| case otlpProtoTrace: | ||
| receiver.tracesUnmarshaler = &ptrace.ProtoUnmarshaler{} | ||
| default: | ||
| return fmt.Errorf("cannot start receiver: build in encoding %s is not supported for traces", receiver.config.Encoding) | ||
| } | ||
| } | ||
| if receiver.logsConsumer != nil { | ||
| switch encodingID { | ||
| case otlpProtoLog: | ||
| receiver.logsUnmarshaler = &plog.ProtoUnmarshaler{} | ||
| case rawTextLog: | ||
| receiver.logsUnmarshaler = unmarshalLogStrings{} | ||
| case cloudLogging: | ||
| receiver.logsUnmarshaler = unmarshalCloudLoggingLogEntry{} | ||
| default: | ||
| return fmt.Errorf("cannot start receiver: build in encoding %s is not supported for logs", receiver.config.Encoding) | ||
| } | ||
| } | ||
| if receiver.metricsConsumer != nil { | ||
| switch encodingID { | ||
| case otlpProtoMetric: | ||
| receiver.metricsUnmarshaler = &pmetric.ProtoUnmarshaler{} | ||
| default: | ||
| return fmt.Errorf("cannot start receiver: build in encoding %s is not supported for metrics", receiver.config.Encoding) | ||
| } | ||
| } | ||
| } | ||
| createHandlerFn = receiver.createReceiverHandler | ||
| } else { | ||
| // we will rely on the attributes of the message to determine the signal, so we need all proto unmarshalers | ||
| receiver.tracesUnmarshaler = &ptrace.ProtoUnmarshaler{} | ||
| receiver.metricsUnmarshaler = &pmetric.ProtoUnmarshaler{} | ||
| receiver.logsUnmarshaler = &plog.ProtoUnmarshaler{} | ||
| createHandlerFn = receiver.createMultiplexingReceiverHandler | ||
| } |
There was a problem hiding this comment.
There are a lot of branching conditions here, can you break it up into some helper functions.
There was a problem hiding this comment.
Pushed, update... but I have the impression there is something wrong with the CI/CD. Waiting until that is resolved.
There was a problem hiding this comment.
@TylerHelmuth I've split it into two functions (mainly grouping built-in and external encoders).
TylerHelmuth
removed
the
ready to merge
alexvanboxel
force-pushed
the
pubsubreceiver-encodingextensions
branch
from
98cca94 to
7512333
Compare
alexvanboxel
force-pushed
the
pubsubreceiver-encodingextensions
branch
from
06d3e3c to
c9f46c8
Compare
|
Re-merged to fix the merge conflict. |
atoulme
approved these changes
Jan 22, 2025
atoulme
added
the
ready to merge
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Added support for encoding extensions. Setting the encoding field in the config now references the extension. If it didn't find the extension, it will fall back to searching the internal encoders.
To make the build in encoders consistent with the extensions, they now have the same interface.
Link to tracking issue
Fixes #37109
Testing
Documentation
Reworked the encoding section of the README, with an example of a text_encoding extension