@@ -5,25 +5,72 @@ package internal // import "go.opentelemetry.io/collector/exporter/exporterhelpe
55
66import (
77 "context"
8+ "errors"
89
910 "go.uber.org/zap"
1011
12+ "go.opentelemetry.io/collector/component"
1113 "go.opentelemetry.io/collector/exporter/exporterbatcher"
1214 "go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
1315 "go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
1416 "go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
15- "go.opentelemetry.io/collector/exporter/exporterqueue"
1617)
1718
1819// QueueBatchSettings is a subset of the queuebatch.Settings that are needed when used within an Exporter.
1920type QueueBatchSettings [K any ] struct {
20- Encoding exporterqueue .Encoding [K ]
21+ Encoding queuebatch .Encoding [K ]
2122 Sizers map [exporterbatcher.SizerType ]queuebatch.Sizer [K ]
2223}
2324
25+ // NewDefaultQueueConfig returns the default config for QueueConfig.
26+ func NewDefaultQueueConfig () QueueConfig {
27+ return QueueConfig {
28+ Enabled : true ,
29+ NumConsumers : 10 ,
30+ // By default, batches are 8192 spans, for a total of up to 8 million spans in the queue
31+ // This can be estimated at 1-4 GB worth of maximum memory usage
32+ // This default is probably still too high, and may be adjusted further down in a future release
33+ QueueSize : 1_000 ,
34+ Blocking : false ,
35+ }
36+ }
37+
38+ // QueueConfig defines configuration for queueing requests before exporting.
39+ // It's supposed to be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.
40+ // Experimental: This API is at the early stage of development and may change without backward compatibility
41+ // until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.
42+ type QueueConfig struct {
43+ // Enabled indicates whether to not enqueue batches before exporting.
44+ Enabled bool `mapstructure:"enabled"`
45+ // NumConsumers is the number of consumers from the queue.
46+ NumConsumers int `mapstructure:"num_consumers"`
47+ // QueueSize is the maximum number of requests allowed in queue at any given time.
48+ QueueSize int `mapstructure:"queue_size"`
49+ // Blocking controls the queue behavior when full.
50+ // If true it blocks until enough space to add the new request to the queue.
51+ Blocking bool `mapstructure:"blocking"`
52+ // StorageID if not empty, enables the persistent storage and uses the component specified
53+ // as a storage extension for the persistent queue
54+ StorageID * component.ID `mapstructure:"storage"`
55+ }
56+
57+ // Validate checks if the Config is valid
58+ func (qCfg * QueueConfig ) Validate () error {
59+ if ! qCfg .Enabled {
60+ return nil
61+ }
62+ if qCfg .NumConsumers <= 0 {
63+ return errors .New ("`num_consumers` must be positive" )
64+ }
65+ if qCfg .QueueSize <= 0 {
66+ return errors .New ("`queue_size` must be positive" )
67+ }
68+ return nil
69+ }
70+
2471func NewQueueSender (
2572 qSet queuebatch.Settings [request.Request ],
26- qCfg exporterqueue. Config ,
73+ qCfg QueueConfig ,
2774 bCfg exporterbatcher.Config ,
2875 exportFailureMessage string ,
2976 next sender.Sender [request.Request ],
@@ -43,7 +90,7 @@ func NewQueueSender(
4390 return queuebatch .NewQueueBatch (qSet , newQueueBatchConfig (qCfg , bCfg ), exportFunc )
4491}
4592
46- func newQueueBatchConfig (qCfg exporterqueue. Config , bCfg exporterbatcher.Config ) queuebatch.Config {
93+ func newQueueBatchConfig (qCfg QueueConfig , bCfg exporterbatcher.Config ) queuebatch.Config {
4794 qbCfg := queuebatch.Config {
4895 Enabled : true ,
4996 WaitForResult : ! qCfg .Enabled ,
0 commit comments