DSET-3874: Add retry configuration similar to OTel (#17)

* DSET-3874: Add retry configuration similar to OTel

* Remove client.worker - it's not needed

* Update Release Notes

Author: martin-majlis-s1

RELEASE_NOTES.md

@@ -1,8 +1,13 @@
 # Release Notes
 
-## 0.0.6 Fix Concurency Issues
+## 0.0.7 Add more retry parameters
 
-* OpenTelemetry can call AddEvents multiple times in parallel. Lets use another Pub/Sub to publish events into topic and then consume them independently.
+* To make OpenTelemetry configuration more stable we have introduced more retry options. We have to propagate them to the library.
+* Fix another data race caused by `client.worker`.
+
+## 0.0.6 Fix Concurrency Issues
+
+* OpenTelemetry can call AddEvents multiple times in parallel. Let's use another Pub/Sub to publish events into topic and then consume them independently.
 
 ## 0.0.5 Quick Hack
 

examples/client/main.go

@@ -21,6 +21,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/buffer_config"
+
 	"github.com/scalyr/dataset-go/pkg/api/add_events"
 	"github.com/scalyr/dataset-go/pkg/client"
 	"github.com/scalyr/dataset-go/pkg/config"
@@ -41,7 +43,14 @@ func main() {
 	}
 
 	// manually adjust delay between sending buffers
-	cfg, err = cfg.Update(config.WithMaxBufferDelay(3 * BatchDelay))
+	bufCfg, err := cfg.BufferSettings.Update(buffer_config.WithMaxLifetime(3 * BatchDelay))
+	if err != nil {
+		panic(err)
+	}
+	cfg, err = cfg.Update(config.WithBufferSettings(*bufCfg))
+	if err != nil {
+		panic(err)
+	}
 
 	// build client
 	cl, err := client.NewClient(

pkg/buffer/buffer.go

@@ -360,8 +360,8 @@ func (buffer *Buffer) ShouldSendSize() bool {
 	return buffer.countEvents.Load() > 0 && buffer.BufferLengths() > ShouldSentBufferSize
 }
 
-func (buffer *Buffer) ShouldSendAge(delay time.Duration) bool {
-	return buffer.countEvents.Load() > 0 && time.Since(time.Unix(0, buffer.createdAt.Load())) > delay
+func (buffer *Buffer) ShouldSendAge(lifetime time.Duration) bool {
+	return buffer.countEvents.Load() > 0 && time.Since(time.Unix(0, buffer.createdAt.Load())) > lifetime
 }
 
 func (buffer *Buffer) BufferLengths() int32 {

pkg/buffer_config/buffer_config.go

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2023 SentinelOne, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package buffer_config
+
+import (
+	"time"
+)
+
+type DataSetBufferSettings struct {
+	MaxLifetime          time.Duration
+	MaxSize              int
+	GroupBy              []string
+	RetryInitialInterval time.Duration
+	RetryMaxInterval     time.Duration
+	RetryMaxElapsedTime  time.Duration
+}
+
+type DataSetBufferSettingsOption func(*DataSetBufferSettings) error
+
+func WithMaxLifetime(maxLifetime time.Duration) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.MaxLifetime = maxLifetime
+		return nil
+	}
+}
+
+func WithMaxSize(maxSize int) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.MaxSize = maxSize
+		return nil
+	}
+}
+
+func WithGroupBy(groupBy []string) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.GroupBy = groupBy
+		return nil
+	}
+}
+
+func WithRetryInitialInterval(retryInitialInterval time.Duration) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.RetryInitialInterval = retryInitialInterval
+		return nil
+	}
+}
+
+func WithRetryMaxInterval(retryMaxInterval time.Duration) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.RetryMaxInterval = retryMaxInterval
+		return nil
+	}
+}
+
+func WithRetryMaxElapsedTime(retryMaxElapsedTime time.Duration) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.RetryMaxElapsedTime = retryMaxElapsedTime
+		return nil
+	}
+}
+
+func New(opts ...DataSetBufferSettingsOption) (*DataSetBufferSettings, error) {
+	cfg := &DataSetBufferSettings{}
+	for _, opt := range opts {
+		if err := opt(cfg); err != nil {
+			return nil, err
+		}
+	}
+	return cfg, nil
+}
+
+func (cfg *DataSetBufferSettings) Update(opts ...DataSetBufferSettingsOption) (*DataSetBufferSettings, error) {
+	newCfg := *cfg
+	for _, opt := range opts {
+		if err := opt(&newCfg); err != nil {
+			return &newCfg, err
+		}
+	}
+	return &newCfg, nil
+}

pkg/buffer_config/buffer_config_test.go

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2023 SentinelOne, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package buffer_config
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/maxatome/go-testdeep/helpers/tdsuite"
+	"github.com/maxatome/go-testdeep/td"
+)
+
+type SuiteBufferSettings struct{}
+
+func (s *SuiteBufferSettings) PreTest(t *td.T, testName string) error {
+	os.Clearenv()
+	return nil
+}
+
+func (s *SuiteBufferSettings) PostTest(t *td.T, testName string) error {
+	os.Clearenv()
+	return nil
+}
+
+func (s *SuiteBufferSettings) Destroy(t *td.T) error {
+	os.Clearenv()
+	return nil
+}
+
+func TestSuiteBufferSettings(t *testing.T) {
+	td.NewT(t)
+	tdsuite.Run(t, &SuiteBufferSettings{})
+}
+
+func (s *SuiteBufferSettings) TestConfigWithOptions(assert, require *td.T) {
+	bufCfg, errB := New(
+		WithMaxLifetime(3*time.Second),
+		WithMaxSize(12345),
+		WithGroupBy([]string{"aaa", "bbb"}),
+		WithRetryInitialInterval(8*time.Second),
+		WithRetryMaxInterval(30*time.Second),
+		WithRetryMaxElapsedTime(10*time.Minute),
+	)
+
+	assert.CmpNoError(errB)
+
+	assert.Cmp(*bufCfg, DataSetBufferSettings{
+		MaxLifetime:          3 * time.Second,
+		MaxSize:              12345,
+		GroupBy:              []string{"aaa", "bbb"},
+		RetryInitialInterval: 8 * time.Second,
+		RetryMaxInterval:     30 * time.Second,
+		RetryMaxElapsedTime:  10 * time.Minute,
+	})
+}
+
+func (s *SuiteBufferSettings) TestDataConfigUpdate(assert, require *td.T) {
+	bufCfg, errB := New(
+		WithMaxLifetime(3*time.Second),
+		WithMaxSize(12345),
+		WithGroupBy([]string{"aaa", "bbb"}),
+		WithRetryInitialInterval(8*time.Second),
+		WithRetryMaxInterval(30*time.Second),
+		WithRetryMaxElapsedTime(10*time.Minute),
+	)
+	assert.CmpNoError(errB)
+
+	assert.Cmp(*bufCfg, DataSetBufferSettings{
+		MaxLifetime:          3 * time.Second,
+		MaxSize:              12345,
+		GroupBy:              []string{"aaa", "bbb"},
+		RetryInitialInterval: 8 * time.Second,
+		RetryMaxInterval:     30 * time.Second,
+		RetryMaxElapsedTime:  10 * time.Minute,
+	})
+
+	bufCfg2, err := bufCfg.Update(
+		WithMaxLifetime(23*time.Second),
+		WithMaxSize(212345),
+		WithGroupBy([]string{"2aaa", "2bbb"}),
+		WithRetryInitialInterval(28*time.Second),
+		WithRetryMaxInterval(230*time.Second),
+		WithRetryMaxElapsedTime(210*time.Minute),
+	)
+	assert.CmpNoError(err)
+
+	// original config is unchanged
+	assert.Cmp(*bufCfg, DataSetBufferSettings{
+		MaxLifetime:          3 * time.Second,
+		MaxSize:              12345,
+		GroupBy:              []string{"aaa", "bbb"},
+		RetryInitialInterval: 8 * time.Second,
+		RetryMaxInterval:     30 * time.Second,
+		RetryMaxElapsedTime:  10 * time.Minute,
+	})
+
+	// new config is changed
+	assert.Cmp(*bufCfg2, DataSetBufferSettings{
+		MaxLifetime:          23 * time.Second,
+		MaxSize:              212345,
+		GroupBy:              []string{"2aaa", "2bbb"},
+		RetryInitialInterval: 28 * time.Second,
+		RetryMaxInterval:     230 * time.Second,
+		RetryMaxElapsedTime:  210 * time.Minute,
+	})
+}

pkg/client/add_events.go

@@ -53,7 +53,7 @@ func (client *DataSetClient) AddEvents(bundles []*add_events.EventBundle) error
 	// first, figure out which keys are part of the batch
 	seenKeys := make(map[string]bool)
 	for _, bundle := range bundles {
-		key := bundle.Key(client.Config.GroupBy)
+		key := bundle.Key(client.Config.BufferSettings.GroupBy)
 		seenKeys[key] = true
 	}
 
@@ -73,7 +73,7 @@ func (client *DataSetClient) AddEvents(bundles []*add_events.EventBundle) error
 
 	// and as last step - publish them
 	for _, bundle := range bundles {
-		key := bundle.Key(client.Config.GroupBy)
+		key := bundle.Key(client.Config.BufferSettings.GroupBy)
 		client.eventsEnqueued.Add(1)
 		client.addEventsPubSub.Pub(bundle, key)
 	}
@@ -172,13 +172,13 @@ func (client *DataSetClient) ListenAndSendBundlesForKey(key string, ch chan inte
 	}
 }
 
-// IsProcessingBuffers returns True if there are still some unprocessed data.
+// IsProcessingBuffers returns True if there are still some unprocessed buffers.
 // False otherwise.
 func (client *DataSetClient) IsProcessingBuffers() bool {
 	return client.buffersEnqueued.Load() > client.buffersProcessed.Load()
 }
 
-// IsProcessingBuffers returns True if there are still some unprocessed data.
+// IsProcessingEvents returns True if there are still some unprocessed events.
 // False otherwise.
 func (client *DataSetClient) IsProcessingEvents() bool {
 	return client.eventsEnqueued.Load() > client.eventsProcessed.Load()
@@ -198,7 +198,7 @@ func (client *DataSetClient) Finish() {
 			zap.Uint64("eventsEnqueued", client.eventsEnqueued.Load()),
 			zap.Uint64("eventsProcessed", client.eventsProcessed.Load()),
 		)
-		time.Sleep(client.Config.RetryBase)
+		time.Sleep(client.Config.BufferSettings.RetryInitialInterval)
 		i++
 		if i > 50 {
 			break
@@ -216,13 +216,12 @@ func (client *DataSetClient) Finish() {
 			zap.Uint64("buffersEnqueued", client.buffersEnqueued.Load()),
 			zap.Uint64("buffersProcessed", client.buffersProcessed.Load()),
 		)
-		time.Sleep(client.Config.RetryBase)
+		time.Sleep(client.Config.BufferSettings.RetryInitialInterval)
 		j++
 		if j > 50 {
 			break
 		}
 	}
-	client.workers.Wait()
 
 	client.Logger.Info("All buffers have been processed")
 }

pkg/client/add_events_long_running_test.go

@@ -28,6 +28,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/buffer_config"
+
 	"github.com/maxatome/go-testdeep/helpers/tdsuite"
 
 	"github.com/scalyr/dataset-go/pkg/api/add_events"
@@ -65,11 +67,13 @@ func TestSuiteAddEventsLongRunning(t *testing.T) {
 func (s *SuiteAddEventsLongRunning) TestAddEventsManyLogsShouldSucceed(assert, require *td.T) {
 	const MaxDelayMs = 200
 	config := &config.DataSetConfig{
-		Endpoint:       "https://example.com",
-		Tokens:         config.DataSetTokens{WriteLog: "AAAA"},
-		MaxPayloadB:    1000,
-		MaxBufferDelay: time.Duration(MaxDelayMs) * time.Millisecond,
-		RetryBase:      RetryBase,
+		Endpoint: "https://example.com",
+		Tokens:   config.DataSetTokens{WriteLog: "AAAA"},
+		BufferSettings: buffer_config.DataSetBufferSettings{
+			MaxSize:              1000,
+			MaxLifetime:          time.Duration(MaxDelayMs) * time.Millisecond,
+			RetryInitialInterval: RetryBase,
+		},
 	}
 	sc, _ := NewClient(config, &http.Client{}, zap.Must(zap.NewDevelopment()))