go/opentelemetry-go
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-go.git synced 2025-11-27 22:49:15 +02:00
Code Issues Releases Activity

Default implementation for empty BatchProcessor (#5239)

Browse Source 
Ensure an empty BatchProcessor does not panic when any method is called.
Default an empty BatchProcessor as being shut down.
This commit is contained in:
Tyler Yahn
2024-04-22 07:21:49 -07:00
committed by GitHub
parent 9370c5a01f
commit b34cfc47c4
2 changed files with 19 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
sdk/log/batch.go
View File

@@ -29,7 +29,9 @@ const (
var _ Processor = (*BatchProcessor)(nil) var _ Processor = (*BatchProcessor)(nil)
// BatchProcessor is a processor that exports batches of log records. // BatchProcessor is a processor that exports batches of log records.
// A BatchProcessor must be created with [NewBatchProcessor]. //
// Use [NewBatchProcessor] to create a BatchProcessor. An empty BatchProcessor
// is shut down by default, no records will be batched or exported.
type BatchProcessor struct { type BatchProcessor struct {
// The BatchProcessor 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
@@ -170,7 +172,7 @@ func (b *BatchProcessor) poll(interval time.Duration) (done chan struct{}) {
// OnEmit batches provided log record. // OnEmit batches provided log record.
func (b *BatchProcessor) OnEmit(_ context.Context, r Record) error { func (b *BatchProcessor) OnEmit(_ context.Context, r Record) error {
if b.stopped.Load() { if b.stopped.Load() || b.q == nil {
return nil return nil
} }
if n := b.q.Enqueue(r); n >= b.batchSize { if n := b.q.Enqueue(r); n >= b.batchSize {
@@ -187,12 +189,12 @@ func (b *BatchProcessor) OnEmit(_ context.Context, r Record) error {
// Enabled returns if b is enabled. // Enabled returns if b is enabled.
func (b *BatchProcessor) Enabled(context.Context, Record) bool { func (b *BatchProcessor) Enabled(context.Context, Record) bool {
return !b.stopped.Load() return !b.stopped.Load() && b.q != nil
} }
// Shutdown flushes queued log records and shuts down the decorated exporter. // Shutdown flushes queued log records and shuts down the decorated exporter.
func (b *BatchProcessor) Shutdown(ctx context.Context) error { func (b *BatchProcessor) Shutdown(ctx context.Context) error {
if b.stopped.Swap(true) { if b.stopped.Swap(true) || b.q == nil {
return nil return nil
} }
@@ -219,7 +221,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 *BatchProcessor) ForceFlush(ctx context.Context) error { func (b *BatchProcessor) ForceFlush(ctx context.Context) error {
if b.stopped.Load() { if b.stopped.Load() || b.q == nil {
return nil return nil
} }

12
sdk/log/batch_test.go
View File

@@ -18,6 +18,18 @@ import (
"go.opentelemetry.io/otel/log" "go.opentelemetry.io/otel/log"
) )
func TestEmptyBatchConfig(t *testing.T) {
assert.NotPanics(t, func() {
var bp BatchProcessor
ctx := context.Background()
var record Record
assert.NoError(t, bp.OnEmit(ctx, record), "OnEmit")
assert.False(t, bp.Enabled(ctx, record), "Enabled")
assert.NoError(t, bp.ForceFlush(ctx), "ForceFlush")
assert.NoError(t, bp.Shutdown(ctx), "Shutdown")
})
}
func TestNewBatchConfig(t *testing.T) { func TestNewBatchConfig(t *testing.T) {
otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) { otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) {
t.Log(err) t.Log(err)