69 lines
2.4 KiB
Go
69 lines
2.4 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package batchprocessor // import "go.opentelemetry.io/collector/processor/batchprocessor"
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"go.opentelemetry.io/collector/component"
|
|
)
|
|
|
|
// Config defines configuration for batch processor.
|
|
type Config struct {
|
|
// Timeout sets the time after which a batch will be sent regardless of size.
|
|
// When this is set to zero, batched data will be sent immediately.
|
|
Timeout time.Duration `mapstructure:"timeout"`
|
|
|
|
// SendBatchSize is the size of a batch which after hit, will trigger it to be sent.
|
|
// When this is set to zero, the batch size is ignored and data will be sent immediately
|
|
// subject to only send_batch_max_size.
|
|
SendBatchSize uint32 `mapstructure:"send_batch_size"`
|
|
|
|
// SendBatchMaxSize is the maximum size of a batch. It must be larger than SendBatchSize.
|
|
// Larger batches are split into smaller units.
|
|
// Default value is 0, that means no maximum size.
|
|
SendBatchMaxSize uint32 `mapstructure:"send_batch_max_size"`
|
|
|
|
// MetadataKeys is a list of client.Metadata keys that will be
|
|
// used to form distinct batchers. If this setting is empty,
|
|
// a single batcher instance will be used. When this setting
|
|
// is not empty, one batcher will be used per distinct
|
|
// combination of values for the listed metadata keys.
|
|
//
|
|
// Empty value and unset metadata are treated as distinct cases.
|
|
//
|
|
// Entries are case-insensitive. Duplicated entries will
|
|
// trigger a validation error.
|
|
MetadataKeys []string `mapstructure:"metadata_keys"`
|
|
|
|
// MetadataCardinalityLimit indicates the maximum number of
|
|
// batcher instances that will be created through a distinct
|
|
// combination of MetadataKeys.
|
|
MetadataCardinalityLimit uint32 `mapstructure:"metadata_cardinality_limit"`
|
|
}
|
|
|
|
var _ component.Config = (*Config)(nil)
|
|
|
|
// Validate checks if the processor configuration is valid
|
|
func (cfg *Config) Validate() error {
|
|
if cfg.SendBatchMaxSize > 0 && cfg.SendBatchMaxSize < cfg.SendBatchSize {
|
|
return errors.New("send_batch_max_size must be greater or equal to send_batch_size")
|
|
}
|
|
uniq := map[string]bool{}
|
|
for _, k := range cfg.MetadataKeys {
|
|
l := strings.ToLower(k)
|
|
if _, has := uniq[l]; has {
|
|
return fmt.Errorf("duplicate entry in metadata_keys: %q (case-insensitive)", l)
|
|
}
|
|
uniq[l] = true
|
|
}
|
|
if cfg.Timeout < 0 {
|
|
return errors.New("timeout must be greater or equal to 0")
|
|
}
|
|
return nil
|
|
}
|