try to avoid race in backoff (#584)

ozontech · Feb 5, 2024 · 5a3d4d1 · 5a3d4d1
1 parent 77e9347
commit 5a3d4d1
Show file tree

Hide file tree

Showing 7 changed files with 38 additions and 22 deletions.
diff --git a/pipeline/backoff.go b/pipeline/backoff.go
@@ -9,8 +9,8 @@ import (
 
 type RetriableBatcher struct {
 	outFn        RetriableBatcherOutFn
-	backoff      backoff.BackOff
 	batcher      *Batcher
+	backoffOpts  BackoffOpts
 	onRetryError func(err error)
 }
 
@@ -19,19 +19,13 @@ type RetriableBatcherOutFn func(*WorkerData, *Batch) error
 type BackoffOpts struct {
 	MinRetention time.Duration
 	Multiplier   float64
-	AttemptNum   uint64
+	AttemptNum   int
 }
 
 func NewRetriableBatcher(batcherOpts *BatcherOptions, batcherOutFn RetriableBatcherOutFn, opts BackoffOpts, onError func(err error)) *RetriableBatcher {
-	boff := GetBackoff(
-		opts.MinRetention,
-		opts.Multiplier,
-		opts.AttemptNum,
-	)
-
 	batcherBackoff := &RetriableBatcher{
 		outFn:        batcherOutFn,
-		backoff:      boff,
+		backoffOpts:  opts,
 		onRetryError: onError,
 	}
 	batcherBackoff.setBatcher(batcherOpts)
@@ -44,14 +38,36 @@ func (b *RetriableBatcher) setBatcher(batcherOpts *BatcherOptions) {
 }
 
 func (b *RetriableBatcher) Out(data *WorkerData, batch *Batch) {
-	b.backoff.Reset()
-
-	err := backoff.Retry(func() error {
-		return b.outFn(data, batch)
-	}, b.backoff)
+	exponentionalBackoff := backoff.ExponentialBackOff{
+		InitialInterval:     b.backoffOpts.MinRetention,
+		Multiplier:          b.backoffOpts.Multiplier,
+		RandomizationFactor: 0.5,
+		MaxInterval:         backoff.DefaultMaxInterval,
+		MaxElapsedTime:      backoff.DefaultMaxElapsedTime,
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}
+	exponentionalBackoff.Reset()
 
-	if err != nil {
-		b.onRetryError(err)
+	var timer *time.Timer
+	numTries := 0
+	for {
+		err := b.outFn(data, batch)
+		if err == nil {
+			return
+		}
+		next := exponentionalBackoff.NextBackOff()
+		if next == backoff.Stop || (b.backoffOpts.AttemptNum >= 0 && numTries > b.backoffOpts.AttemptNum) {
+			b.onRetryError(err)
+			return
+		}
+		numTries++
+		if timer == nil {
+			timer = time.NewTimer(next)
+		} else {
+			timer.Reset(next)
+		}
+		<-timer.C
 	}
 }
 

diff --git a/plugin/output/clickhouse/clickhouse.go b/plugin/output/clickhouse/clickhouse.go
@@ -428,7 +428,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {

diff --git a/plugin/output/elasticsearch/elasticsearch.go b/plugin/output/elasticsearch/elasticsearch.go
@@ -243,7 +243,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {

diff --git a/plugin/output/gelf/gelf.go b/plugin/output/gelf/gelf.go
@@ -230,7 +230,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {

diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go
@@ -195,7 +195,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {

diff --git a/plugin/output/postgres/postgres.go b/plugin/output/postgres/postgres.go
@@ -289,7 +289,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {

diff --git a/plugin/output/splunk/splunk.go b/plugin/output/splunk/splunk.go
@@ -146,7 +146,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	backoffOpts := pipeline.BackoffOpts{
 		MinRetention: p.config.Retention_,
 		Multiplier:   float64(p.config.RetentionExponentMultiplier),
-		AttemptNum:   uint64(p.config.Retry),
+		AttemptNum:   p.config.Retry,
 	}
 
 	onError := func(err error) {