You've already forked opentelemetry-go
mirror of
https://github.com/open-telemetry/opentelemetry-go.git
synced 2025-11-27 22:49:15 +02:00
sdk/log: Rename BatchingProcessor to BatchProcessor (#5229)
This commit is contained in:
@@ -55,10 +55,10 @@ The user can configure custom processors and decorate built-in processors.
|
|||||||
The [Simple processor](https://opentelemetry.io/docs/specs/otel/logs/sdk/#simple-processor)
|
The [Simple processor](https://opentelemetry.io/docs/specs/otel/logs/sdk/#simple-processor)
|
||||||
is implemented as `SimpleProcessor` struct in [simple.go](simple.go).
|
is implemented as `SimpleProcessor` struct in [simple.go](simple.go).
|
||||||
|
|
||||||
### BatchingProcessor
|
### BatchProcessor
|
||||||
|
|
||||||
The [Batching processor](https://opentelemetry.io/docs/specs/otel/logs/sdk/#batching-processor)
|
The [Batching processor](https://opentelemetry.io/docs/specs/otel/logs/sdk/#batching-processor)
|
||||||
is implemented as `BatchingProcessor` struct in [batch.go](batch.go).
|
is implemented as `BatchProcessor` struct in [batch.go](batch.go).
|
||||||
|
|
||||||
The `Batcher` can be also configured using the `OTEL_BLRP_*` environment variables as
|
The `Batcher` can be also configured using the `OTEL_BLRP_*` environment variables as
|
||||||
[defined by the specification](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#batch-logrecord-processor).
|
[defined by the specification](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#batch-logrecord-processor).
|
||||||
|
|||||||
@@ -25,20 +25,20 @@ const (
|
|||||||
envarExpMaxBatchSize = "OTEL_BLRP_MAX_EXPORT_BATCH_SIZE"
|
envarExpMaxBatchSize = "OTEL_BLRP_MAX_EXPORT_BATCH_SIZE"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Compile-time check BatchingProcessor implements Processor.
|
// Compile-time check BatchProcessor implements Processor.
|
||||||
var _ Processor = (*BatchingProcessor)(nil)
|
var _ Processor = (*BatchProcessor)(nil)
|
||||||
|
|
||||||
// BatchingProcessor is a processor that exports batches of log records.
|
// BatchProcessor is a processor that exports batches of log records.
|
||||||
// A BatchingProcessor must be created with [NewBatchingProcessor].
|
// A BatchProcessor must be created with [NewBatchProcessor].
|
||||||
type BatchingProcessor struct {
|
type BatchProcessor struct {
|
||||||
// The BatchingProcessor is designed to provide the highest throughput of
|
// The BatchProcessor is designed to provide the highest throughput of
|
||||||
// log records possible while being compatible with OpenTelemetry. The
|
// log records possible while being compatible with OpenTelemetry. The
|
||||||
// entry point of log records is the OnEmit method. This method is designed
|
// entry point of log records is the OnEmit method. This method is designed
|
||||||
// to receive records as fast as possible while still honoring shutdown
|
// to receive records as fast as possible while still honoring shutdown
|
||||||
// commands. All records received are enqueued to queue.
|
// commands. All records received are enqueued to queue.
|
||||||
//
|
//
|
||||||
// In order to block OnEmit as little as possible, a separate "poll"
|
// In order to block OnEmit as little as possible, a separate "poll"
|
||||||
// goroutine is spawned at the creation of a BatchingProcessor. This
|
// goroutine is spawned at the creation of a BatchProcessor. This
|
||||||
// goroutine is responsible for batching the queue at regular polled
|
// goroutine is responsible for batching the queue at regular polled
|
||||||
// intervals, or when it is directly signaled to.
|
// intervals, or when it is directly signaled to.
|
||||||
//
|
//
|
||||||
@@ -48,7 +48,7 @@ type BatchingProcessor struct {
|
|||||||
// separate goroutine dedicated to the export. This asynchronous behavior
|
// separate goroutine dedicated to the export. This asynchronous behavior
|
||||||
// allows the poll goroutine to maintain accurate interval polling.
|
// allows the poll goroutine to maintain accurate interval polling.
|
||||||
//
|
//
|
||||||
// __BatchingProcessor__ __Poll Goroutine__ __Export Goroutine__
|
// __BatchProcessor__ __Poll Goroutine__ __Export Goroutine__
|
||||||
// || || || || || ||
|
// || || || || || ||
|
||||||
// || ********** || || || || ********** ||
|
// || ********** || || || || ********** ||
|
||||||
// || Records=>* OnEmit * || || | - ticker || || * export * ||
|
// || Records=>* OnEmit * || || | - ticker || || * export * ||
|
||||||
@@ -91,15 +91,15 @@ type BatchingProcessor struct {
|
|||||||
// pollDone signals the poll goroutine has completed.
|
// pollDone signals the poll goroutine has completed.
|
||||||
pollDone chan struct{}
|
pollDone chan struct{}
|
||||||
|
|
||||||
// stopped holds the stopped state of the BatchingProcessor.
|
// stopped holds the stopped state of the BatchProcessor.
|
||||||
stopped atomic.Bool
|
stopped atomic.Bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewBatchingProcessor decorates the provided exporter
|
// NewBatchProcessor decorates the provided exporter
|
||||||
// so that the log records are batched before exporting.
|
// so that the log records are batched before exporting.
|
||||||
//
|
//
|
||||||
// All of the exporter's methods are called synchronously.
|
// All of the exporter's methods are called synchronously.
|
||||||
func NewBatchingProcessor(exporter Exporter, opts ...BatchProcessorOption) *BatchingProcessor {
|
func NewBatchProcessor(exporter Exporter, opts ...BatchProcessorOption) *BatchProcessor {
|
||||||
cfg := newBatchingConfig(opts)
|
cfg := newBatchingConfig(opts)
|
||||||
if exporter == nil {
|
if exporter == nil {
|
||||||
// Do not panic on nil export.
|
// Do not panic on nil export.
|
||||||
@@ -113,7 +113,7 @@ func NewBatchingProcessor(exporter Exporter, opts ...BatchProcessorOption) *Batc
|
|||||||
// appropriately on export.
|
// appropriately on export.
|
||||||
exporter = newChunkExporter(exporter, cfg.expMaxBatchSize.Value)
|
exporter = newChunkExporter(exporter, cfg.expMaxBatchSize.Value)
|
||||||
|
|
||||||
b := &BatchingProcessor{
|
b := &BatchProcessor{
|
||||||
// TODO: explore making the size of this configurable.
|
// TODO: explore making the size of this configurable.
|
||||||
exporter: newBufferExporter(exporter, 1),
|
exporter: newBufferExporter(exporter, 1),
|
||||||
|
|
||||||
@@ -128,7 +128,7 @@ func NewBatchingProcessor(exporter Exporter, opts ...BatchProcessorOption) *Batc
|
|||||||
|
|
||||||
// poll spawns a goroutine to handle interval polling and batch exporting. The
|
// poll spawns a goroutine to handle interval polling and batch exporting. The
|
||||||
// returned done chan is closed when the spawned goroutine completes.
|
// returned done chan is closed when the spawned goroutine completes.
|
||||||
func (b *BatchingProcessor) poll(interval time.Duration) (done chan struct{}) {
|
func (b *BatchProcessor) poll(interval time.Duration) (done chan struct{}) {
|
||||||
done = make(chan struct{})
|
done = make(chan struct{})
|
||||||
|
|
||||||
ticker := time.NewTicker(interval)
|
ticker := time.NewTicker(interval)
|
||||||
@@ -169,7 +169,7 @@ func (b *BatchingProcessor) poll(interval time.Duration) (done chan struct{}) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// OnEmit batches provided log record.
|
// OnEmit batches provided log record.
|
||||||
func (b *BatchingProcessor) OnEmit(_ context.Context, r Record) error {
|
func (b *BatchProcessor) OnEmit(_ context.Context, r Record) error {
|
||||||
if b.stopped.Load() {
|
if b.stopped.Load() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -186,12 +186,12 @@ func (b *BatchingProcessor) OnEmit(_ context.Context, r Record) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Enabled returns if b is enabled.
|
// Enabled returns if b is enabled.
|
||||||
func (b *BatchingProcessor) Enabled(context.Context, Record) bool {
|
func (b *BatchProcessor) Enabled(context.Context, Record) bool {
|
||||||
return !b.stopped.Load()
|
return !b.stopped.Load()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Shutdown flushes queued log records and shuts down the decorated exporter.
|
// Shutdown flushes queued log records and shuts down the decorated exporter.
|
||||||
func (b *BatchingProcessor) Shutdown(ctx context.Context) error {
|
func (b *BatchProcessor) Shutdown(ctx context.Context) error {
|
||||||
if b.stopped.Swap(true) {
|
if b.stopped.Swap(true) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -218,7 +218,7 @@ var ctxErr = func(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// ForceFlush flushes queued log records and flushes the decorated exporter.
|
// ForceFlush flushes queued log records and flushes the decorated exporter.
|
||||||
func (b *BatchingProcessor) ForceFlush(ctx context.Context) error {
|
func (b *BatchProcessor) ForceFlush(ctx context.Context) error {
|
||||||
if b.stopped.Load() {
|
if b.stopped.Load() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -372,7 +372,7 @@ func newBatchingConfig(options []BatchProcessorOption) batchingConfig {
|
|||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
|
|
||||||
// BatchProcessorOption applies a configuration to a [BatchingProcessor].
|
// BatchProcessorOption applies a configuration to a [BatchProcessor].
|
||||||
type BatchProcessorOption interface {
|
type BatchProcessorOption interface {
|
||||||
apply(batchingConfig) batchingConfig
|
apply(batchingConfig) batchingConfig
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -145,17 +145,17 @@ func TestNewBatchingConfig(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestBatchingProcessor(t *testing.T) {
|
func TestBatchProcessor(t *testing.T) {
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
t.Run("NilExporter", func(t *testing.T) {
|
t.Run("NilExporter", func(t *testing.T) {
|
||||||
assert.NotPanics(t, func() { NewBatchingProcessor(nil) })
|
assert.NotPanics(t, func() { NewBatchProcessor(nil) })
|
||||||
})
|
})
|
||||||
|
|
||||||
t.Run("Polling", func(t *testing.T) {
|
t.Run("Polling", func(t *testing.T) {
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
const size = 15
|
const size = 15
|
||||||
b := NewBatchingProcessor(
|
b := NewBatchProcessor(
|
||||||
e,
|
e,
|
||||||
WithMaxQueueSize(2*size),
|
WithMaxQueueSize(2*size),
|
||||||
WithExportMaxBatchSize(2*size),
|
WithExportMaxBatchSize(2*size),
|
||||||
@@ -178,7 +178,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
t.Run("OnEmit", func(t *testing.T) {
|
t.Run("OnEmit", func(t *testing.T) {
|
||||||
const batch = 10
|
const batch = 10
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
b := NewBatchingProcessor(
|
b := NewBatchProcessor(
|
||||||
e,
|
e,
|
||||||
WithMaxQueueSize(10*batch),
|
WithMaxQueueSize(10*batch),
|
||||||
WithExportMaxBatchSize(batch),
|
WithExportMaxBatchSize(batch),
|
||||||
@@ -201,7 +201,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
e.ExportTrigger = make(chan struct{})
|
e.ExportTrigger = make(chan struct{})
|
||||||
|
|
||||||
const batch = 10
|
const batch = 10
|
||||||
b := NewBatchingProcessor(
|
b := NewBatchProcessor(
|
||||||
e,
|
e,
|
||||||
WithMaxQueueSize(3*batch),
|
WithMaxQueueSize(3*batch),
|
||||||
WithExportMaxBatchSize(batch),
|
WithExportMaxBatchSize(batch),
|
||||||
@@ -236,7 +236,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
t.Run("Enabled", func(t *testing.T) {
|
t.Run("Enabled", func(t *testing.T) {
|
||||||
b := NewBatchingProcessor(defaultNoopExporter)
|
b := NewBatchProcessor(defaultNoopExporter)
|
||||||
assert.True(t, b.Enabled(ctx, Record{}))
|
assert.True(t, b.Enabled(ctx, Record{}))
|
||||||
|
|
||||||
_ = b.Shutdown(ctx)
|
_ = b.Shutdown(ctx)
|
||||||
@@ -246,14 +246,14 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
t.Run("Shutdown", func(t *testing.T) {
|
t.Run("Shutdown", func(t *testing.T) {
|
||||||
t.Run("Error", func(t *testing.T) {
|
t.Run("Error", func(t *testing.T) {
|
||||||
e := newTestExporter(assert.AnError)
|
e := newTestExporter(assert.AnError)
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
assert.ErrorIs(t, b.Shutdown(ctx), assert.AnError, "exporter error not returned")
|
assert.ErrorIs(t, b.Shutdown(ctx), assert.AnError, "exporter error not returned")
|
||||||
assert.NoError(t, b.Shutdown(ctx))
|
assert.NoError(t, b.Shutdown(ctx))
|
||||||
})
|
})
|
||||||
|
|
||||||
t.Run("Multiple", func(t *testing.T) {
|
t.Run("Multiple", func(t *testing.T) {
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
|
|
||||||
const shutdowns = 3
|
const shutdowns = 3
|
||||||
for i := 0; i < shutdowns; i++ {
|
for i := 0; i < shutdowns; i++ {
|
||||||
@@ -264,7 +264,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
|
|
||||||
t.Run("OnEmit", func(t *testing.T) {
|
t.Run("OnEmit", func(t *testing.T) {
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
assert.NoError(t, b.Shutdown(ctx))
|
assert.NoError(t, b.Shutdown(ctx))
|
||||||
|
|
||||||
want := e.ExportN()
|
want := e.ExportN()
|
||||||
@@ -274,7 +274,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
|
|
||||||
t.Run("ForceFlush", func(t *testing.T) {
|
t.Run("ForceFlush", func(t *testing.T) {
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
|
|
||||||
assert.NoError(t, b.OnEmit(ctx, Record{}))
|
assert.NoError(t, b.OnEmit(ctx, Record{}))
|
||||||
assert.NoError(t, b.Shutdown(ctx))
|
assert.NoError(t, b.Shutdown(ctx))
|
||||||
@@ -287,7 +287,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
e.ExportTrigger = make(chan struct{})
|
e.ExportTrigger = make(chan struct{})
|
||||||
t.Cleanup(func() { close(e.ExportTrigger) })
|
t.Cleanup(func() { close(e.ExportTrigger) })
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
c, cancel := context.WithCancel(ctx)
|
c, cancel := context.WithCancel(ctx)
|
||||||
@@ -300,7 +300,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
t.Run("ForceFlush", func(t *testing.T) {
|
t.Run("ForceFlush", func(t *testing.T) {
|
||||||
t.Run("Flush", func(t *testing.T) {
|
t.Run("Flush", func(t *testing.T) {
|
||||||
e := newTestExporter(assert.AnError)
|
e := newTestExporter(assert.AnError)
|
||||||
b := NewBatchingProcessor(
|
b := NewBatchProcessor(
|
||||||
e,
|
e,
|
||||||
WithMaxQueueSize(100),
|
WithMaxQueueSize(100),
|
||||||
WithExportMaxBatchSize(10),
|
WithExportMaxBatchSize(10),
|
||||||
@@ -336,7 +336,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
t.Cleanup(func() { ctxErr = orig })
|
t.Cleanup(func() { ctxErr = orig })
|
||||||
|
|
||||||
const batch = 1
|
const batch = 1
|
||||||
b := NewBatchingProcessor(
|
b := NewBatchProcessor(
|
||||||
e,
|
e,
|
||||||
WithMaxQueueSize(10*batch),
|
WithMaxQueueSize(10*batch),
|
||||||
WithExportMaxBatchSize(batch),
|
WithExportMaxBatchSize(batch),
|
||||||
@@ -385,7 +385,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
t.Run("CanceledContext", func(t *testing.T) {
|
t.Run("CanceledContext", func(t *testing.T) {
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
e.ExportTrigger = make(chan struct{})
|
e.ExportTrigger = make(chan struct{})
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
t.Cleanup(func() { _ = b.Shutdown(ctx) })
|
t.Cleanup(func() { _ = b.Shutdown(ctx) })
|
||||||
|
|
||||||
var r Record
|
var r Record
|
||||||
@@ -404,7 +404,7 @@ func TestBatchingProcessor(t *testing.T) {
|
|||||||
const goRoutines = 10
|
const goRoutines = 10
|
||||||
|
|
||||||
e := newTestExporter(nil)
|
e := newTestExporter(nil)
|
||||||
b := NewBatchingProcessor(e)
|
b := NewBatchProcessor(e)
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
|||||||
@@ -192,7 +192,7 @@ func WithResource(res *resource.Resource) LoggerProviderOption {
|
|||||||
// Each WithProcessor creates a separate pipeline. Use custom decorators
|
// Each WithProcessor creates a separate pipeline. Use custom decorators
|
||||||
// for advanced scenarios such as enriching with attributes.
|
// for advanced scenarios such as enriching with attributes.
|
||||||
//
|
//
|
||||||
// For production, use [NewBatchingProcessor] to batch log records before they are exported.
|
// For production, use [NewBatchProcessor] to batch log records before they are exported.
|
||||||
// For testing and debugging, use [NewSimpleProcessor] to synchronously export log records.
|
// For testing and debugging, use [NewSimpleProcessor] to synchronously export log records.
|
||||||
func WithProcessor(processor Processor) LoggerProviderOption {
|
func WithProcessor(processor Processor) LoggerProviderOption {
|
||||||
return loggerProviderOptionFunc(func(cfg providerConfig) providerConfig {
|
return loggerProviderOptionFunc(func(cfg providerConfig) providerConfig {
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ type SimpleProcessor struct {
|
|||||||
// This Processor is not recommended for production use. The synchronous
|
// This Processor is not recommended for production use. The synchronous
|
||||||
// nature of this Processor make it good for testing, debugging, or
|
// nature of this Processor make it good for testing, debugging, or
|
||||||
// showing examples of other features, but it can be slow and have a high
|
// showing examples of other features, but it can be slow and have a high
|
||||||
// computation resource usage overhead. [NewBatchingProcessor] is recommended
|
// computation resource usage overhead. [NewBatchProcessor] is recommended
|
||||||
// for production use instead.
|
// for production use instead.
|
||||||
func NewSimpleProcessor(exporter Exporter, _ ...SimpleProcessorOption) *SimpleProcessor {
|
func NewSimpleProcessor(exporter Exporter, _ ...SimpleProcessorOption) *SimpleProcessor {
|
||||||
if exporter == nil {
|
if exporter == nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user