DSET-4050 feat: make client shutdown timeout configurable. (#44)

* DSET-4050 fix indefinite retrials while shutdown of client

- do not reset ExponentialBackOff
- define separate timeout for shutdown

* add serverHost getter

* release 0.0.11

Author: zdaratom-s1

RELEASE_NOTES.md

@@ -1,5 +1,9 @@
 # Release Notes
 
+## 0.0.11
+
+* feat: make client shutdown timeout configurable.
+
 ## 0.0.10
 
 * feat: make `serverHost` explicit field of the `Event`.

pkg/buffer_config/buffer_config.go

@@ -43,6 +43,7 @@ type DataSetBufferSettings struct {
 	RetryInitialInterval     time.Duration
 	RetryMaxInterval         time.Duration
 	RetryMaxElapsedTime      time.Duration
+	RetryShutdownTimeout     time.Duration // defines timeout period (for which client will retry on failures) for processing events and sending buffers after shutdown of a client
 }
 
 func NewDefaultDataSetBufferSettings() DataSetBufferSettings {
@@ -55,6 +56,7 @@ func NewDefaultDataSetBufferSettings() DataSetBufferSettings {
 		RetryMaxElapsedTime:      300 * time.Second,
 		RetryRandomizationFactor: backoff.DefaultRandomizationFactor,
 		RetryMultiplier:          backoff.DefaultMultiplier,
+		RetryShutdownTimeout:     30 * time.Second,
 	}
 }
 
@@ -116,6 +118,13 @@ func WithRetryMaxElapsedTime(retryMaxElapsedTime time.Duration) DataSetBufferSet
 	}
 }
 
+func WithRetryShutdownTimeout(retryShutdownTimeout time.Duration) DataSetBufferSettingsOption {
+	return func(c *DataSetBufferSettings) error {
+		c.RetryShutdownTimeout = retryShutdownTimeout
+		return nil
+	}
+}
+
 func New(opts ...DataSetBufferSettingsOption) (*DataSetBufferSettings, error) {
 	cfg := &DataSetBufferSettings{}
 	for _, opt := range opts {

pkg/client/add_events.go

@@ -211,8 +211,10 @@ func (client *DataSetClient) isProcessingEvents() bool {
 	return client.eventsEnqueued.Load() > client.eventsProcessed.Load()
 }
 
-// Shutdown stops processing of new events and waits until all the events that are
-// being processed are really processed (sent to DataSet).
+// Shutdown takes care of shutdown of client. It does following steps
+// - stops processing of new events,
+// - tries (with 1st half of shutdownMaxTimeout period) to process (add into buffers) all the events,
+// - tries (with 2nd half of shutdownMaxTimeout period) to send processed events (buffers) into DataSet
 func (client *DataSetClient) Shutdown() error {
 	client.Logger.Info("Shutting down - BEGIN")
 	// mark as finished to prevent processing of further events
@@ -222,33 +224,25 @@ func (client *DataSetClient) Shutdown() error {
 	client.logStatistics()
 
 	var lastError error = nil
+	shutdownTimeout := minDuration(client.Config.BufferSettings.RetryMaxElapsedTime, client.Config.BufferSettings.RetryShutdownTimeout)
 	expBackoff := backoff.ExponentialBackOff{
 		InitialInterval:     client.Config.BufferSettings.RetryInitialInterval,
 		RandomizationFactor: client.Config.BufferSettings.RetryRandomizationFactor,
 		Multiplier:          client.Config.BufferSettings.RetryMultiplier,
 		MaxInterval:         client.Config.BufferSettings.RetryMaxInterval,
-		MaxElapsedTime:      client.Config.BufferSettings.RetryMaxElapsedTime,
+		MaxElapsedTime:      shutdownTimeout / 2,
 		Stop:                backoff.Stop,
 		Clock:               backoff.SystemClock,
 	}
 	expBackoff.Reset()
 
-	// first we wait until all the events in buffers are added into buffers
-	// then we are waiting until all the buffers are processed
-	// if some progress is made we restart the waiting times
-
-	// do wait for all events to be processed
+	// try (with timeout) to process (add into buffers) events,
 	retryNum := 0
-	lastProcessed := client.eventsProcessed.Load()
+	processingStart := time.Now()
 	for client.isProcessingEvents() {
 		// log statistics
 		client.logStatistics()
 
-		// if some events were processed restart retry interval
-		if client.eventsProcessed.Load() != lastProcessed {
-			expBackoff.Reset()
-		}
-		lastProcessed = client.eventsProcessed.Load()
 		backoffDelay := expBackoff.NextBackOff()
 		client.Logger.Info(
 			"Shutting down - processing events",
@@ -279,22 +273,26 @@ func (client *DataSetClient) Shutdown() error {
 	client.Logger.Info("Shutting down - publishing all buffers")
 	client.publishAllBuffers()
 
-	// do wait for all buffers to be processed
+	// reinitialize expBackoff with MaxElapsedTime based on actually elapsed time of processing (previous phase)
+	processingElapsed := time.Since(processingStart)
+	remainingShutdownTimeout := maxDuration(shutdownTimeout-processingElapsed, shutdownTimeout/2)
+	expBackoff = backoff.ExponentialBackOff{
+		InitialInterval:     client.Config.BufferSettings.RetryInitialInterval,
+		RandomizationFactor: client.Config.BufferSettings.RetryRandomizationFactor,
+		Multiplier:          client.Config.BufferSettings.RetryMultiplier,
+		MaxInterval:         client.Config.BufferSettings.RetryMaxInterval,
+		MaxElapsedTime:      remainingShutdownTimeout,
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}
+	// do wait (with timeout) for all buffers to be sent to the server
 	retryNum = 0
 	expBackoff.Reset()
-	lastProcessed = client.buffersProcessed.Load()
-	lastDropped := client.buffersDropped.Load()
-	initialDropped := lastDropped
+	initialDropped := client.buffersDropped.Load()
 	for client.isProcessingBuffers() {
 		// log statistics
 		client.logStatistics()
 
-		// if some buffers were processed restart retry interval
-		if client.buffersProcessed.Load()+lastDropped != lastProcessed+client.buffersDropped.Load() {
-			expBackoff.Reset()
-		}
-		lastProcessed = client.buffersProcessed.Load()
-		lastDropped = client.buffersDropped.Load()
 		backoffDelay := expBackoff.NextBackOff()
 		client.Logger.Info(
 			"Shutting down - processing buffers",
@@ -476,3 +474,17 @@ func truncateText(text string, length int) string {
 
 	return text
 }
+
+func minDuration(a, b time.Duration) time.Duration {
+	if a <= b {
+		return a
+	}
+	return b
+}
+
+func maxDuration(a, b time.Duration) time.Duration {
+	if a >= b {
+		return a
+	}
+	return b
+}

pkg/client/add_events_test.go

@@ -582,7 +582,8 @@ func TestAddEventsLogResponseBodyOnInvalidJson(t *testing.T) {
 	server := mockServer(t, 503, []byte("<html>not valid json</html>"))
 	defer server.Close()
 	config := newDataSetConfig(server.URL, *newBufferSettings(
-		buffer_config.WithRetryMaxElapsedTime(time.Duration(3) * time.Second),
+		buffer_config.WithRetryMaxElapsedTime(time.Duration(30)*time.Second),
+		buffer_config.WithRetryShutdownTimeout(time.Duration(6)*time.Second),
 	), server_host_config.NewDefaultDataSetServerHostSettings())
 	sc, err := NewClient(config, &http.Client{}, zap.Must(zap.NewDevelopment()), nil)
 	require.Nil(t, err)
@@ -605,6 +606,39 @@ func TestAddEventsLogResponseBodyOnInvalidJson(t *testing.T) {
 	assert.GreaterOrEqual(t, attempt.Load(), int32(0))
 }
 
+func TestShutdownFinishesWithinExpectedTimeout(t *testing.T) {
+	// GIVEN
+	attempt.Store(0)
+	server := mockServer(t, http.StatusTooManyRequests, []byte("{}"))
+	defer server.Close()
+	retryShutdownTimeout := 6
+	config := newDataSetConfig(server.URL, *newBufferSettings(
+		buffer_config.WithRetryMaxElapsedTime(time.Duration(30)*time.Second),
+		buffer_config.WithRetryShutdownTimeout(time.Duration(retryShutdownTimeout)*time.Second),
+	), server_host_config.NewDefaultDataSetServerHostSettings())
+	sc, err := NewClient(config, &http.Client{}, zap.Must(zap.NewDevelopment()), nil)
+	require.Nil(t, err)
+
+	sessionInfo := &add_events.SessionInfo{ServerId: "a", ServerType: "b"}
+	sc.SessionInfo = sessionInfo
+	event1 := &add_events.Event{Thread: "5", Sev: 3, Ts: "0", Attrs: map[string]interface{}{"message": "test - 1"}}
+	eventBundle1 := &add_events.EventBundle{Event: event1, Thread: &add_events.Thread{Id: "5", Name: "fred"}}
+	err = sc.AddEvents([]*add_events.EventBundle{eventBundle1})
+	assert.Nil(t, err)
+	// WHEN
+	shutdownStart := time.Now()
+	err = sc.Shutdown()
+	shutdownFinish := time.Since(shutdownStart)
+	// THEN
+	assert.LessOrEqual(t, shutdownFinish, time.Duration(retryShutdownTimeout+1)*time.Second)
+	lastError := sc.LastError()
+	assert.Nil(t, lastError)
+
+	assert.NotNil(t, err)
+	assert.Errorf(t, err, "some buffers were dropped during finishing - 1")
+	assert.GreaterOrEqual(t, attempt.Load(), int32(0))
+}
+
 func TestAddEventsAreNotRejectedOncePreviousReqRetriesMaxLifetimeExpired(t *testing.T) {
 	// GIVEN
 	maxElapsedTime := 10

pkg/client/client.go

@@ -665,3 +665,7 @@ func (client *DataSetClient) lastErrorTimestamp() time.Time {
 func (client *DataSetClient) isLastRetryableErrorTimedOut() bool {
 	return client.lastErrorTimestamp().Add(client.Config.BufferSettings.RetryMaxElapsedTime).Before(time.Now())
 }
+
+func (client *DataSetClient) ServerHost() string {
+	return client.serverHost
+}

pkg/version/version.go

@@ -17,6 +17,6 @@
 package version
 
 const (
-	Version      = "0.0.10"
-	ReleasedDate = "2023-07-10"
+	Version      = "0.0.11"
+	ReleasedDate = "2023-07-26"
 )

pkg/version/version_test.go

@@ -23,6 +23,6 @@ import (
 )
 
 func TestVersion(t *testing.T) {
-	assert.Equal(t, "0.0.10", Version)
-	assert.Equal(t, "2023-07-10", ReleasedDate)
+	assert.Equal(t, "0.0.11", Version)
+	assert.Equal(t, "2023-07-26", ReleasedDate)
 }