Skip to content

Commit 8bfb7af

Browse files
axwaxw
committed
Use new exporterbatcher.SizeConfig
See open-telemetry/opentelemetry-collector#12486
1 parent b2d5487 commit 8bfb7af

File tree

6 files changed

+105
-22
lines changed

6 files changed

+105
-22
lines changed

‎.chloggen/elasticsearchexporter-sizeconfig.yaml

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
# Use this changelog template to create an entry for release notes.
2+
3+
# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
4+
change_type: deprecation
5+
6+
# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
7+
component: elasticsearchexporter
8+
9+
# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
10+
note: Deprecate `batcher::min_size_items` and `batcher::max_size_items` in favor of `batcher::min_size` and `batcher::max_size`.
11+
12+
# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
13+
issues: [38243]
14+
15+
# (Optional) One or more lines of additional information to render under the primary note.
16+
# These lines will be padded with 2 spaces and then inserted directly into the document.
17+
# Use pipe (|) for multiline entries.
18+
subtext:
19+
20+
# If your change doesn't affect end users or the exported elements of any package,
21+
# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
22+
# Optional: The change log or logs in which this entry should be included.
23+
# e.g. '[user]' or '[user, api]'
24+
# Include 'user' if the change is relevant to end users.
25+
# Include 'api' if there is a change to a library API.
26+
# Default: '[user]'
27+
change_logs: [user]

‎exporter/elasticsearchexporter/README.md

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,8 +94,11 @@ The Elasticsearch exporter supports the [common `batcher` settings](https://gith
9494

9595
- `batcher`:
9696
- `enabled` (default=unset): Enable batching of requests into 1 or more bulk requests. On a batcher flush, it is possible for a batched request to be translated to more than 1 bulk request due to `flush::bytes`.
97-
- `min_size_items` (default=5000): Minimum number of log records / spans / data points in the batched request to immediately trigger a batcher flush.
98-
- `max_size_items` (default=0): Maximum number of log records / spans / data points in a batched request. To limit bulk request size, configure `flush::bytes` instead. :warning: It is recommended to keep `max_size_items` as 0 as a non-zero value may lead to broken metrics grouping and indexing rejections.
97+
- `sizer` (default=items): Unit of `min_size` and `max_size`. Currently supports only "items", in the future will also support "bytes".
98+
- `min_size` (default=5000): Minimum batch size to be exported to Elasticsearch, measured in units according to `batcher::sizer`.
99+
- `max_size` (default=0): Maximum batch size to be exported to Elasticsearch, measured in units according to `batcher::sizer`. To limit bulk request size, configure `flush::bytes` instead. :warning: It is recommended to keep `max_size` as 0 as a non-zero value may lead to broken metrics grouping and indexing rejections.
100+
- `min_size_items` (DEPRECATED, use `batcher::min_size` instead): Minimum number of log records / spans / data points in the batched request to immediately trigger a batcher flush.
101+
- `max_size_items` (DEPRECATED, use `batcher::max_size` instead): Maximum number of log records / spans / data points in a batched request.
99102
- `flush_timeout` (default=30s): Maximum time of the oldest item spent inside the batcher buffer, aka "max age of batcher buffer". A batcher flush will happen regardless of the size of content in batcher buffer.
100103

101104
By default, the exporter will perform its own buffering and batching, as configured through the

‎exporter/elasticsearchexporter/config.go

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"go.opentelemetry.io/collector/config/configcompression"
1616
"go.opentelemetry.io/collector/config/confighttp"
1717
"go.opentelemetry.io/collector/config/configopaque"
18+
"go.opentelemetry.io/collector/confmap"
1819
"go.opentelemetry.io/collector/exporter/exporterbatcher"
1920
"go.opentelemetry.io/collector/exporter/exporterhelper"
2021
"go.uber.org/zap"
@@ -96,10 +97,31 @@ type BatcherConfig struct {
9697
// FlushTimeout sets the time after which a batch will be sent regardless of its size.
9798
FlushTimeout time.Duration `mapstructure:"flush_timeout"`
9899

100+
// SizeConfig sets the size limits for a batch.
101+
exporterbatcher.SizeConfig `mapstructure:",squash"`
102+
103+
// Deprecated: [v0.121.0] Ignored if SizeConfig is set.
99104
exporterbatcher.MinSizeConfig `mapstructure:",squash"`
105+
// Deprecated: [v0.121.0] Ignored if SizeConfig is set.
100106
exporterbatcher.MaxSizeConfig `mapstructure:",squash"`
101107
}
102108

109+
func (c *BatcherConfig) Unmarshal(conf *confmap.Conf) error {
110+
if err := conf.Unmarshal(c); err != nil {
111+
return err
112+
}
113+
114+
if c.MinSizeItems != nil && !conf.IsSet("min_size") {
115+
c.SizeConfig.MinSize = *c.MinSizeItems
116+
}
117+
118+
if c.MaxSizeItems != nil && !conf.IsSet("max_size") {
119+
c.SizeConfig.MaxSize = *c.MaxSizeItems
120+
}
121+
122+
return nil
123+
}
124+
103125
type TelemetrySettings struct {
104126
LogRequestBody bool `mapstructure:"log_request_body"`
105127
LogResponseBody bool `mapstructure:"log_response_body"`

‎exporter/elasticsearchexporter/config_test.go

Lines changed: 36 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -118,11 +118,9 @@ func TestConfig(t *testing.T) {
118118
},
119119
Batcher: BatcherConfig{
120120
FlushTimeout: 30 * time.Second,
121-
MinSizeConfig: exporterbatcher.MinSizeConfig{ //nolint:staticcheck
122-
MinSizeItems: &defaultBatcherMinSizeItems,
123-
},
124-
MaxSizeConfig: exporterbatcher.MaxSizeConfig{ //nolint:staticcheck
125-
MaxSizeItems: nil,
121+
SizeConfig: exporterbatcher.SizeConfig{
122+
Sizer: exporterbatcher.SizerTypeItems,
123+
MinSize: defaultBatcherMinSizeItems,
126124
},
127125
},
128126
},
@@ -192,11 +190,9 @@ func TestConfig(t *testing.T) {
192190
},
193191
Batcher: BatcherConfig{
194192
FlushTimeout: 30 * time.Second,
195-
MinSizeConfig: exporterbatcher.MinSizeConfig{ //nolint:staticcheck
196-
MinSizeItems: &defaultBatcherMinSizeItems,
197-
},
198-
MaxSizeConfig: exporterbatcher.MaxSizeConfig{ //nolint:staticcheck
199-
MaxSizeItems: nil,
193+
SizeConfig: exporterbatcher.SizeConfig{
194+
Sizer: exporterbatcher.SizerTypeItems,
195+
MinSize: defaultBatcherMinSizeItems,
200196
},
201197
},
202198
},
@@ -266,11 +262,9 @@ func TestConfig(t *testing.T) {
266262
},
267263
Batcher: BatcherConfig{
268264
FlushTimeout: 30 * time.Second,
269-
MinSizeConfig: exporterbatcher.MinSizeConfig{ //nolint:staticcheck
270-
MinSizeItems: &defaultBatcherMinSizeItems,
271-
},
272-
MaxSizeConfig: exporterbatcher.MaxSizeConfig{ //nolint:staticcheck
273-
MaxSizeItems: nil,
265+
SizeConfig: exporterbatcher.SizeConfig{
266+
Sizer: exporterbatcher.SizerTypeItems,
267+
MinSize: defaultBatcherMinSizeItems,
274268
},
275269
},
276270
},
@@ -327,6 +321,33 @@ func TestConfig(t *testing.T) {
327321
cfg.Compression = "gzip"
328322
}),
329323
},
324+
{
325+
id: component.NewIDWithName(metadata.Type, "batcher_minmax_size_items"),
326+
configFile: "config.yaml",
327+
expected: withDefaultConfig(func(cfg *Config) {
328+
cfg.Endpoint = "https://elastic.example.com:9200"
329+
330+
cfg.Batcher.MinSize = 100
331+
cfg.Batcher.MaxSize = 200
332+
cfg.Batcher.MinSizeItems = &cfg.Batcher.MinSize
333+
cfg.Batcher.MaxSizeItems = &cfg.Batcher.MaxSize
334+
}),
335+
},
336+
{
337+
id: component.NewIDWithName(metadata.Type, "batcher_minmax_size"),
338+
configFile: "config.yaml",
339+
expected: withDefaultConfig(func(cfg *Config) {
340+
cfg.Endpoint = "https://elastic.example.com:9200"
341+
342+
cfg.Batcher.MinSize = 100
343+
cfg.Batcher.MaxSize = 200
344+
345+
minSizeItems := 300
346+
maxSizeItems := 400
347+
cfg.Batcher.MinSizeItems = &minSizeItems
348+
cfg.Batcher.MaxSizeItems = &maxSizeItems
349+
}),
350+
},
330351
}
331352

332353
for _, tt := range tests {

‎exporter/elasticsearchexporter/factory.go

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -95,11 +95,9 @@ func createDefaultConfig() component.Config {
9595
},
9696
Batcher: BatcherConfig{
9797
FlushTimeout: 30 * time.Second,
98-
MinSizeConfig: exporterbatcher.MinSizeConfig{ //nolint:staticcheck
99-
MinSizeItems: &defaultBatcherMinSizeItems,
100-
},
101-
MaxSizeConfig: exporterbatcher.MaxSizeConfig{ //nolint:staticcheck
102-
MaxSizeItems: nil,
98+
SizeConfig: exporterbatcher.SizeConfig{
99+
Sizer: exporterbatcher.SizerTypeItems,
100+
MinSize: defaultBatcherMinSizeItems,
103101
},
104102
},
105103
Flush: FlushSettings{

‎exporter/elasticsearchexporter/testdata/config.yaml

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,3 +89,15 @@ elasticsearch/compression_none:
8989
elasticsearch/compression_gzip:
9090
endpoint: https://elastic.example.com:9200
9191
compression: gzip
92+
elasticsearch/batcher_minmax_size_items:
93+
endpoint: https://elastic.example.com:9200
94+
batcher:
95+
min_size_items: 100
96+
max_size_items: 200
97+
elasticsearch/batcher_minmax_size:
98+
endpoint: https://elastic.example.com:9200
99+
batcher:
100+
min_size: 100
101+
max_size: 200
102+
min_size_items: 300 # min_size is set, does not carry over
103+
max_size_items: 400 # max_size is set, does not carry over

0 commit comments

Comments
 (0)