mirror of
https://github.com/hanzoai/telemetry.git
synced 2026-10-30 20:01:49 +00:00
1da69f7208
<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue. Ex. Adding a feature - Explain what this achieves.--> #### Description <!-- Issue number if applicable --> Uses `configoptional.Optional` for `exporterhelper.QueueBatchConfig`. #### Link to tracking issue Updates #14021 --------- Co-authored-by: Evan Bradley <[email protected]> Co-authored-by: Joshua MacDonald <[email protected]>
59 lines
1.8 KiB
Go
59 lines
1.8 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package internal // import "go.opentelemetry.io/collector/exporter/exporterhelper/internal"
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"go.opentelemetry.io/collector/config/configoptional"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/queuebatch"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/request"
|
|
"go.opentelemetry.io/collector/exporter/exporterhelper/internal/sender"
|
|
)
|
|
|
|
// NewDefaultQueueConfig returns the default config for queuebatch.Config.
|
|
// By default:
|
|
//
|
|
// - the queue stores 1000 requests of telemetry
|
|
// - is non-blocking when full
|
|
// - concurrent exports limited to 10
|
|
// - emits batches of 8192 items, timeout 200ms
|
|
func NewDefaultQueueConfig() queuebatch.Config {
|
|
return queuebatch.Config{
|
|
Sizer: request.SizerTypeRequests,
|
|
NumConsumers: 10,
|
|
QueueSize: 1_000,
|
|
BlockOnOverflow: false,
|
|
Batch: configoptional.Default(queuebatch.BatchConfig{
|
|
FlushTimeout: 200 * time.Millisecond,
|
|
Sizer: request.SizerTypeItems,
|
|
MinSize: 8192,
|
|
}),
|
|
}
|
|
}
|
|
|
|
func NewQueueSender(
|
|
qSet queuebatch.AllSettings[request.Request],
|
|
qCfg queuebatch.Config,
|
|
exportFailureMessage string,
|
|
next sender.Sender[request.Request],
|
|
) (sender.Sender[request.Request], error) {
|
|
exportFunc := func(ctx context.Context, req request.Request) error {
|
|
// Have to read the number of items before sending the request since the request can
|
|
// be modified by the downstream components like the batcher.
|
|
itemsCount := req.ItemsCount()
|
|
if errSend := next.Send(ctx, req); errSend != nil {
|
|
qSet.Telemetry.Logger.Error("Exporting failed. Dropping data."+exportFailureMessage,
|
|
zap.Error(errSend), zap.Int("dropped_items", itemsCount))
|
|
return errSend
|
|
}
|
|
return nil
|
|
}
|
|
|
|
return queuebatch.NewQueueBatch(qSet, qCfg, exportFunc)
|
|
}
|