[exporter/loadbalancing] Add top level sending_queue, retry_on_failure and timeout settings #36094
+282
−87
There are no files selected for viewing
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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: loadbalancingexporter | ||
|
|
||
| # A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`). | ||
| note: Adding sending_queue, retry_on_failure and timeout settings to loadbalancing exporter configuration | ||
|
|
||
| # Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists. | ||
| issues: [35378,16826] | ||
|
|
||
| # (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: OTLP sub-exporter queue will be automatically disabled if loadbalancing exporter queue is enabled to avoid data loss | ||
|
|
||
| # 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] | ||
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,14 +7,22 @@ package loadbalancingexporter // import "github.com/open-telemetry/opentelemetry | |
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "go.opentelemetry.io/collector/component" | ||
| "go.opentelemetry.io/collector/config/configretry" | ||
| "go.opentelemetry.io/collector/exporter" | ||
| "go.opentelemetry.io/collector/exporter/exporterhelper" | ||
| "go.opentelemetry.io/collector/exporter/otlpexporter" | ||
| "go.uber.org/zap" | ||
|
|
||
| "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/loadbalancingexporter/internal/metadata" | ||
| ) | ||
|
|
||
| const ( | ||
| zapEndpointKey = "endpoint" | ||
| ) | ||
|
|
||
| // NewFactory creates a factory for the exporter. | ||
| func NewFactory() exporter.Factory { | ||
| return exporter.NewFactory( | ||
|
|
@@ -32,20 +40,98 @@ func createDefaultConfig() component.Config { | |
| otlpDefaultCfg.Endpoint = "placeholder:4317" | ||
|
|
||
| return &Config{ | ||
| TimeoutSettings: exporterhelper.NewDefaultTimeoutConfig(), | ||
|
There was a problem hiding this comment. This behavior should be clearly documented. I think it would even make sense to log a warning in case users tried to use that option. That said, there ARE users relying on this feature at the moment. What should we do about them? I think we should copy their current values to the load-balancer level, which would give them roughly the same desired behavior they have today. So, here's how it could work:
I don't think we need a feature flag or deprecation plan for this. |
||
| QueueSettings: exporterhelper.NewDefaultQueueConfig(), | ||
| BackOffConfig: configretry.NewDefaultBackOffConfig(), | ||
| Protocol: Protocol{ | ||
| OTLP: *otlpDefaultCfg, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func buildExporterConfig(cfg *Config, endpoint string) otlpexporter.Config { | ||
| oCfg := cfg.Protocol.OTLP | ||
| oCfg.Endpoint = endpoint | ||
| // If top level queue is enabled - per-endpoint queue must be disable | ||
| // This helps us to avoid unexpected issues with mixing 2 level of exporter queues | ||
| if cfg.QueueSettings.Enabled { | ||
| oCfg.QueueConfig.Enabled = false | ||
|
There was a problem hiding this comment. Looking at the for exp, td := range exporterSegregatedTraces {
start := time.Now()
err := exp.ConsumeTraces(ctx, td)
exp.consumeWG.Done()
errs = multierr.Append(errs, err)
...
}Now that the What might be necessary is an There was a problem hiding this comment. Switching to an export queue also means that includes the queue workers which allow parallel processing, which allows a layer of latency hiding. Including a new "queue" worker in this could be hazardous in how it is done, say if you create a routine for each split batch you could end up with high amount of routine scheduling which can cause a significant performance hit. Then if you do more an async send on channels you have the original risk that this solves due to the queue moving from the current queue structure, to now a buffered channel that now needs to be shared again. For this iteration, I would suggest leaving it out of this change and keep as an issue in the backlog of "investigate and address if required" There was a problem hiding this comment. Missed that part ( |
||
| } | ||
| return oCfg | ||
| } | ||
|
|
||
| func buildExporterSettings(params exporter.Settings, endpoint string) exporter.Settings { | ||
| // Override child exporter ID to segregate metrics from loadbalancing top level | ||
| childName := endpoint | ||
| if params.ID.Name() != "" { | ||
| childName = fmt.Sprintf("%s_%s", params.ID.Name(), childName) | ||
| } | ||
| params.ID = component.NewIDWithName(params.ID.Type(), childName) | ||
| // Add "endpoint" attribute to child exporter logger to segregate logs from loadbalancing top level | ||
| params.Logger = params.Logger.With(zap.String(zapEndpointKey, endpoint)) | ||
|
|
||
| return params | ||
| } | ||
|
|
||
| func createTracesExporter(ctx context.Context, params exporter.Settings, cfg component.Config) (exporter.Traces, error) { | ||
| c := cfg.(*Config) | ||
| exporter, err := newTracesExporter(params, cfg) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("cannot configure loadbalancing traces exporter: %w", err) | ||
| } | ||
|
|
||
| return exporterhelper.NewTraces( | ||
| ctx, | ||
| params, | ||
| cfg, | ||
| exporter.ConsumeTraces, | ||
| exporterhelper.WithStart(exporter.Start), | ||
| exporterhelper.WithShutdown(exporter.Shutdown), | ||
| exporterhelper.WithCapabilities(exporter.Capabilities()), | ||
| exporterhelper.WithTimeout(c.TimeoutSettings), | ||
| exporterhelper.WithQueue(c.QueueSettings), | ||
| exporterhelper.WithRetry(c.BackOffConfig), | ||
| ) | ||
| } | ||
|
|
||
| func createLogsExporter(ctx context.Context, params exporter.Settings, cfg component.Config) (exporter.Logs, error) { | ||
| c := cfg.(*Config) | ||
| exporter, err := newLogsExporter(params, cfg) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("cannot configure loadbalancing logs exporter: %w", err) | ||
| } | ||
|
|
||
| return exporterhelper.NewLogs( | ||
| ctx, | ||
| params, | ||
| cfg, | ||
| exporter.ConsumeLogs, | ||
| exporterhelper.WithStart(exporter.Start), | ||
| exporterhelper.WithShutdown(exporter.Shutdown), | ||
| exporterhelper.WithCapabilities(exporter.Capabilities()), | ||
| exporterhelper.WithTimeout(c.TimeoutSettings), | ||
| exporterhelper.WithQueue(c.QueueSettings), | ||
| exporterhelper.WithRetry(c.BackOffConfig), | ||
| ) | ||
| } | ||
|
|
||
| func createMetricsExporter(ctx context.Context, params exporter.Settings, cfg component.Config) (exporter.Metrics, error) { | ||
| c := cfg.(*Config) | ||
| exporter, err := newMetricsExporter(params, cfg) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("cannot configure loadbalancing metrics exporter: %w", err) | ||
| } | ||
|
|
||
| return exporterhelper.NewMetrics( | ||
| ctx, | ||
| params, | ||
| cfg, | ||
| exporter.ConsumeMetrics, | ||
| exporterhelper.WithStart(exporter.Start), | ||
| exporterhelper.WithShutdown(exporter.Shutdown), | ||
| exporterhelper.WithCapabilities(exporter.Capabilities()), | ||
| exporterhelper.WithTimeout(c.TimeoutSettings), | ||
| exporterhelper.WithQueue(c.QueueSettings), | ||
| exporterhelper.WithRetry(c.BackOffConfig), | ||
| ) | ||
| } | ||
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
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
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
Oops, something went wrong.
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Now that you've consolidated all exporter queues into one, their overall size is smaller. This could lead to dropped data in some deployments, since you are going from
n*queueSizeto1*queueSize- I don't think this is necessarily a breaking change but I think it should be mentioned in the changelog, so users can increase the queue size if they run into issues.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've updated changelog to reflect this part