Skip to content

Commit

Permalink
Remove old push based batcher
Browse files Browse the repository at this point in the history 
Signed-off-by: Bogdan Drutu <[email protected]>
  • Loading branch information
@bogdandrutu
bogdandrutu committed Feb 1, 2025
1 parent 7b671b8 commit 77deac9
Show file tree
Hide file tree
Showing 9 changed files with 581 additions and 1,837 deletions.
25 changes: 25 additions & 0 deletions .chloggen/rm-old-push-based-batcher.yaml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
# Use this changelog template to create an entry for release notes.

# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
change_type: deprecation

# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
component: exporterhelerp

# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
note: Deprecate alpha featuregate "exporter.UsePullingBasedExporterQueueBatcher". Functionality is enabled by default.

# One or more tracking issues or pull requests related to the change
issues: [12233]

# (Optional) One or more lines of additional information to render under the primary note.
# These lines will be padded with 2 spaces and then inserted directly into the document.
# Use pipe (|) for multiline entries.
subtext:

# Optional: The change log or logs in which this entry should be included.
# e.g. '[user]' or '[user, api]'
# Include 'user' if the change is relevant to end users.
# Include 'api' if there is a change to a library API.
# Default: '[user]'
change_logs: [user]
34 changes: 5 additions & 29 deletions exporter/exporterhelper/internal/base_exporter.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,12 @@ import (
"go.opentelemetry.io/collector/pipeline"
)

var usePullingBasedExporterQueueBatcher = featuregate.GlobalRegistry().MustRegister(
var _ = featuregate.GlobalRegistry().MustRegister(
"exporter.UsePullingBasedExporterQueueBatcher",
featuregate.StageBeta,
featuregate.StageDeprecated,
featuregate.WithRegisterFromVersion("v0.115.0"),
featuregate.WithRegisterDescription("if set to true, turns on the pulling-based exporter queue bathcer"),
featuregate.WithRegisterToVersion("v0.119.0"),
featuregate.WithRegisterDescription("If set to true, turns on the pulling-based exporter queue bathcer"),
)

type ObsrepSenderFactory = func(obsrep *ObsReport) Sender[internal.Request]
Expand All @@ -53,7 +54,6 @@ type BaseExporter struct {
// Chain of senders that the exporter helper applies before passing the data to the actual exporter.
// The data is handled by each sender in the respective order starting from the queueSender.
// Most of the senders are optional, and initialized with a no-op path-through sender.
BatchSender Sender[internal.Request]
QueueSender Sender[internal.Request]
ObsrepSender Sender[internal.Request]
RetrySender Sender[internal.Request]
Expand All @@ -73,7 +73,6 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, osf ObsrepSe
}

be := &BaseExporter{
BatchSender: &BaseSender[internal.Request]{},
QueueSender: &BaseSender[internal.Request]{},
ObsrepSender: osf(obsReport),
RetrySender: &BaseSender[internal.Request]{},
Expand Down Expand Up @@ -101,23 +100,8 @@ func NewBaseExporter(set exporter.Settings, signal pipeline.Signal, osf ObsrepSe
be.QueueSender = NewQueueSender(q, be.Set, be.queueCfg.NumConsumers, be.ExportFailureMessage, be.Obsrep, be.BatcherCfg)
}

if !usePullingBasedExporterQueueBatcher.IsEnabled() && be.BatcherCfg.Enabled ||
usePullingBasedExporterQueueBatcher.IsEnabled() && be.BatcherCfg.Enabled && !be.queueCfg.Enabled {
bs := NewBatchSender(be.BatcherCfg, be.Set)
be.BatchSender = bs
}

be.connectSenders()

if bs, ok := be.BatchSender.(*BatchSender); ok {
// If queue sender is enabled assign to the batch sender the same number of workers.
if qs, ok := be.QueueSender.(*QueueSender); ok {
bs.concurrencyLimit = int64(qs.numConsumers)
}
// Batcher sender mutates the data.
be.ConsumerOptions = append(be.ConsumerOptions, consumer.WithCapabilities(consumer.Capabilities{MutatesData: true}))
}

return be, nil
}

Expand All @@ -133,8 +117,7 @@ func (be *BaseExporter) Send(ctx context.Context, req internal.Request) error {

// connectSenders connects the senders in the predefined order.
func (be *BaseExporter) connectSenders() {
be.QueueSender.SetNextSender(be.BatchSender)
be.BatchSender.SetNextSender(be.ObsrepSender)
be.QueueSender.SetNextSender(be.ObsrepSender)
be.ObsrepSender.SetNextSender(be.RetrySender)
be.RetrySender.SetNextSender(be.TimeoutSender)
}
Expand All @@ -145,11 +128,6 @@ func (be *BaseExporter) Start(ctx context.Context, host component.Host) error {
return err
}

// If no error then start the BatchSender.
if err := be.BatchSender.Start(ctx, host); err != nil {
return err
}

// Last start the queueSender.
return be.QueueSender.Start(ctx, host)
}
Expand All @@ -158,8 +136,6 @@ func (be *BaseExporter) Shutdown(ctx context.Context) error {
return multierr.Combine(
// First shutdown the retry sender, so the queue sender can flush the queue without retries.
be.RetrySender.Shutdown(ctx),
// Then shutdown the batch sender
be.BatchSender.Shutdown(ctx),
// Then shutdown the queue sender.
be.QueueSender.Shutdown(ctx),
// Last shutdown the wrapped exporter itself.
Expand Down
102 changes: 37 additions & 65 deletions exporter/exporterhelper/internal/base_exporter_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,80 +38,52 @@ func newNoopObsrepSender(*ObsReport) Sender[internal.Request] {
}

func TestBaseExporter(t *testing.T) {
runTest := func(testName string, enableQueueBatcher bool) {
t.Run(testName, func(t *testing.T) {
defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
be, err := NewBaseExporter(defaultSettings, defaultSignal, newNoopObsrepSender)
require.NoError(t, err)
require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
require.NoError(t, be.Shutdown(context.Background()))
})
}
runTest("enable_queue_batcher", true)
runTest("disable_queue_batcher", false)
be, err := NewBaseExporter(defaultSettings, defaultSignal, newNoopObsrepSender)
require.NoError(t, err)
require.NoError(t, be.Start(context.Background(), componenttest.NewNopHost()))
require.NoError(t, be.Shutdown(context.Background()))
}

func TestBaseExporterWithOptions(t *testing.T) {
runTest := func(testName string, enableQueueBatcher bool) {
t.Run(testName, func(t *testing.T) {
defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
want := errors.New("my error")
be, err := NewBaseExporter(
defaultSettings, defaultSignal, newNoopObsrepSender,
WithStart(func(context.Context, component.Host) error { return want }),
WithShutdown(func(context.Context) error { return want }),
WithTimeout(NewDefaultTimeoutConfig()),
)
require.NoError(t, err)
require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
require.Equal(t, want, be.Shutdown(context.Background()))
})
}
runTest("enable_queue_batcher", true)
runTest("disable_queue_batcher", false)
want := errors.New("my error")
be, err := NewBaseExporter(
defaultSettings, defaultSignal, newNoopObsrepSender,
WithStart(func(context.Context, component.Host) error { return want }),
WithShutdown(func(context.Context) error { return want }),
WithTimeout(NewDefaultTimeoutConfig()),
)
require.NoError(t, err)
require.Equal(t, want, be.Start(context.Background(), componenttest.NewNopHost()))
require.Equal(t, want, be.Shutdown(context.Background()))
}

func TestQueueOptionsWithRequestExporter(t *testing.T) {
runTest := func(testName string, enableQueueBatcher bool) {
t.Run(testName, func(t *testing.T) {
defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
bs, err := NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithRetry(configretry.NewDefaultBackOffConfig()))
require.NoError(t, err)
require.Nil(t, bs.Marshaler)
require.Nil(t, bs.Unmarshaler)
_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
require.Error(t, err)
bs, err := NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithRetry(configretry.NewDefaultBackOffConfig()))
require.NoError(t, err)
require.Nil(t, bs.Marshaler)
require.Nil(t, bs.Unmarshaler)
_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithRetry(configretry.NewDefaultBackOffConfig()), WithQueue(NewDefaultQueueConfig()))
require.Error(t, err)

_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&mockRequest{})),
WithRetry(configretry.NewDefaultBackOffConfig()),
WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[internal.Request]()))
require.Error(t, err)
})
}
runTest("enable_queue_batcher", true)
runTest("disable_queue_batcher", false)
_, err = NewBaseExporter(exportertest.NewNopSettings(), defaultSignal, newNoopObsrepSender,
WithMarshaler(mockRequestMarshaler), WithUnmarshaler(mockRequestUnmarshaler(&mockRequest{})),
WithRetry(configretry.NewDefaultBackOffConfig()),
WithRequestQueue(exporterqueue.NewDefaultConfig(), exporterqueue.NewMemoryQueueFactory[internal.Request]()))
require.Error(t, err)
}

func TestBaseExporterLogging(t *testing.T) {
runTest := func(testName string, enableQueueBatcher bool) {
t.Run(testName, func(t *testing.T) {
defer setFeatureGateForTest(t, usePullingBasedExporterQueueBatcher, enableQueueBatcher)()
set := exportertest.NewNopSettings()
logger, observed := observer.New(zap.DebugLevel)
set.Logger = zap.New(logger)
rCfg := configretry.NewDefaultBackOffConfig()
rCfg.Enabled = false
bs, err := NewBaseExporter(set, defaultSignal, newNoopObsrepSender, WithRetry(rCfg))
require.NoError(t, err)
sendErr := bs.Send(context.Background(), newErrorRequest())
require.Error(t, sendErr)
set := exportertest.NewNopSettings()
logger, observed := observer.New(zap.DebugLevel)
set.Logger = zap.New(logger)
rCfg := configretry.NewDefaultBackOffConfig()
rCfg.Enabled = false
bs, err := NewBaseExporter(set, defaultSignal, newNoopObsrepSender, WithRetry(rCfg))
require.NoError(t, err)
sendErr := bs.Send(context.Background(), newErrorRequest())
require.Error(t, sendErr)

require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 1)
})
}
runTest("enable_queue_batcher", true)
runTest("disable_queue_batcher", false)
require.Len(t, observed.FilterLevelExact(zap.ErrorLevel).All(), 1)
}
Loading

0 comments on commit 77deac9

Please sign in to comment.