Repository: flant/shell-operator
Pull request #407 (2022-06-15 09:01:00)

fix: restore exponential backoff after #388

<!--
Thank you for sending a pull request! Here some tips for contributors:

1. Fill the description template below.
2. Include appropriate tests (if necessary). Make sure that all CI checks passed.
3. If the Pull Request is a work in progress, make use of GitHub's "Draft PR" feature and mark it as such.
-->

#### Overview

Restore exponential backoff for delayed between  failed hooks restarts.
<!-- Describe your changes briefly here. -->

#### What this PR does / why we need it

Failed hook restarts every second after merging #388.
<!--
- Please state in detail why we need this PR and what it solves.
- If your PR closes some of the existing issues, please add links to them here.
  Mentioned issues will be automatically closed.
  Usage: "Closes #<issue number>", or "Closes (paste link of issue)"
-->

#### Special notes for your reviewer

#### Does this PR introduce a user-facing change?

<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->

```release-note

```

diafour: Added tests for exponential delaying and for CancelTaskDelay.

Patch:
diff --git a/pkg/task/queue/task_queue.go b/pkg/task/queue/task_queue.go
index 1f15b5d..d555976 100644
--- a/pkg/task/queue/task_queue.go
+++ b/pkg/task/queue/task_queue.go
@@ -28,10 +28,10 @@ config parameter.
 */
 
 var (
-	WaitLoopCheckInterval = 125 * time.Millisecond
-	DelayOnQueueIsEmpty   = 250 * time.Millisecond
-	DelayOnFailedTask     = 5 * time.Second
-	DelayOnRepeat         = 25 * time.Millisecond
+	DefaultWaitLoopCheckInterval    = 125 * time.Millisecond
+	DefaultDelayOnQueueIsEmpty      = 250 * time.Millisecond
+	DefaultInitialDelayOnFailedTask = 5 * time.Second
+	DefaultDelayOnRepeat            = 25 * time.Millisecond
 )
 
 type TaskStatus string
@@ -76,11 +76,24 @@ type TaskQueue struct {
 
 	measureActionFn     func()
 	measureActionFnOnce sync.Once
+
+	// Timing settings.
+	WaitLoopCheckInterval time.Duration
+	DelayOnQueueIsEmpty   time.Duration
+	DelayOnRepeat         time.Duration
+	ExponentialBackoffFn  func(failureCount int) time.Duration
 }
 
 func NewTasksQueue() *TaskQueue {
 	return &TaskQueue{
 		items: make([]task.Task, 0),
+		// Default timings
+		WaitLoopCheckInterval: DefaultWaitLoopCheckInterval,
+		DelayOnQueueIsEmpty:   DefaultDelayOnQueueIsEmpty,
+		DelayOnRepeat:         DefaultDelayOnRepeat,
+		ExponentialBackoffFn: func(failureCount int) time.Duration {
+			return exponential_backoff.CalculateDelay(DefaultInitialDelayOnFailedTask, failureCount)
+		},
 	}
 }
 
@@ -404,7 +417,7 @@ func (q *TaskQueue) Start() {
 			switch taskRes.Status {
 			case Fail:
 				// Exponential backoff delay before retry.
-				nextSleepDelay = exponential_backoff.CalculateDelay(DelayOnFailedTask, t.GetFailureCount())
+				nextSleepDelay = q.ExponentialBackoffFn(t.GetFailureCount())
 				t.IncrementFailureCount()
 				q.Status = fmt.Sprintf("sleep after fail for %s", nextSleepDelay.String())
 			case Success, Keep:
@@ -431,7 +444,7 @@ func (q *TaskQueue) Start() {
 				q.Status = ""
 			case Repeat:
 				// repeat a current task after a small delay
-				nextSleepDelay = DelayOnRepeat
+				nextSleepDelay = q.DelayOnRepeat
 				q.Status = "repeat head task"
 			}
 
@@ -470,30 +483,33 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 
 	// Initialize wait settings.
 	waitBegin := time.Now()
-	waitUntil := DelayOnQueueIsEmpty
+	waitUntil := q.DelayOnQueueIsEmpty
 	if sleepDelay != 0 {
 		waitUntil = sleepDelay
 	}
 
-	checkTicker := time.NewTicker(WaitLoopCheckInterval)
+	checkTicker := time.NewTicker(q.WaitLoopCheckInterval)
 	q.waitMu.Lock()
 	q.waitInProgress = true
 	q.cancelDelay = false
 	q.waitMu.Unlock()
 
+	origStatus := q.Status
+
 	defer func() {
 		checkTicker.Stop()
 		q.waitMu.Lock()
 		q.waitInProgress = false
 		q.cancelDelay = false
 		q.waitMu.Unlock()
+		q.Status = origStatus
 	}()
 
 	// Wait for the queued task with some delay.
 	// Every tick increases the 'elapsed' counter until it outgrows the waitUntil value.
 	// Or, delay can be canceled to handle new head task immediately.
-	origStatus := q.Status
 	for {
+		checkTask := false
 		select {
 		case <-q.ctx.Done():
 			// Queue is stopped.
@@ -502,36 +518,42 @@ func (q *TaskQueue) waitForTask(sleepDelay time.Duration) task.Task {
 			// Check and update waitUntil.
 			elapsed := time.Since(waitBegin)
 
-			cancelDelay := false
 			q.waitMu.Lock()
 			if q.cancelDelay {
-				cancelDelay = true
+				// Reset waitUntil to check task immediately.
+				waitUntil = elapsed
 			}
 			q.waitMu.Unlock()
 
 			// Wait loop is done or canceled: break select to check for the head task.
-			if elapsed >= waitUntil || cancelDelay {
-				// Increase waitUntil to wait on the next iteration.
-				waitUntil += DelayOnQueueIsEmpty
-				break
+			if elapsed >= waitUntil {
+				// Increase waitUntil to wait on the next iteration and go check for the head task.
+				checkTask = true
 			}
+		}
 
-			// Wait loop still in progress: update queue status.
-			waitSeconds := time.Since(waitBegin).Truncate(time.Second).String()
-			if sleepDelay == 0 {
-				q.Status = fmt.Sprintf("waiting for task %s", waitSeconds)
+		// Break the for-loop to see if the head task can be returned.
+		if checkTask {
+			if q.IsEmpty() {
+				// No task to return: increase wait time.
+				waitUntil += q.DelayOnQueueIsEmpty
 			} else {
-				delay := sleepDelay.Truncate(time.Second).String()
-				q.Status = fmt.Sprintf("%s (%s left of %s delay)", origStatus, waitSeconds, delay)
+				return q.GetFirst()
 			}
 		}
-		if !q.IsEmpty() {
-			return q.GetFirst()
+
+		// Wait loop still in progress: update queue status.
+		waitSeconds := time.Since(waitBegin).Truncate(time.Second).String()
+		if sleepDelay == 0 {
+			q.Status = fmt.Sprintf("waiting for task %s", waitSeconds)
+		} else {
+			delay := sleepDelay.Truncate(time.Second).String()
+			q.Status = fmt.Sprintf("%s (%s left of %s delay)", origStatus, waitSeconds, delay)
 		}
 	}
 }
 
-// CancelTaskDelay breaks wait loop. Useful to break the possible maximum sleep delay.
+// CancelTaskDelay breaks wait loop. Useful to break the possible long sleep delay.
 func (q *TaskQueue) CancelTaskDelay() {
 	q.waitMu.Lock()
 	if q.waitInProgress {


Test patch:
diff --git a/pkg/task/queue/task_queue_test.go b/pkg/task/queue/task_queue_test.go
index 716f602..38bed0b 100644
--- a/pkg/task/queue/task_queue_test.go
+++ b/pkg/task/queue/task_queue_test.go
@@ -2,8 +2,10 @@ package queue
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	. "github.com/onsi/gomega"
 
@@ -72,3 +74,160 @@ func Test_TasksQueue_Remove(t *testing.T) {
 		ContainSubstring("task_03"),
 	))
 }
+
+func Test_ExponentialBackoff(t *testing.T) {
+	g := NewWithT(t)
+	// Init and prefill queue.
+	q := NewTasksQueue()
+	q.WithContext(context.TODO())
+	q.WithName("test-queue")
+	// Since we don't want the test to run for too long, we don't
+	// want to use lengthy times.
+	q.WaitLoopCheckInterval = 5 * time.Millisecond // default is 125ms
+	q.DelayOnQueueIsEmpty = 5 * time.Millisecond   // default is 250ms
+	q.DelayOnRepeat = 5 * time.Millisecond         // default is 25ms
+	// Add one task.
+	Task := &task.BaseTask{Id: "First one"}
+	q.AddFirst(Task)
+
+	// Set handler to fail 10 times and catch timestamps for each task execution.
+	runsAt := make([]time.Time, 0)
+	failureCounts := make([]int, 0)
+	const fails = 10
+	failsCount := fails
+	queueStopCh := make(chan struct{}, 1)
+	q.WithHandler(func(t task.Task) (res TaskResult) {
+		runsAt = append(runsAt, time.Now())
+		failureCounts = append(failureCounts, t.GetFailureCount())
+		if failsCount > 0 {
+			res.Status = Fail
+			failsCount--
+			return
+		}
+		res.Status = Success
+		res.AfterHandle = func() {
+			close(queueStopCh)
+		}
+		return
+	})
+
+	// Set exponential backoff to the constant delay just to wait more than DelayOnQueueIsEmpty.
+	// It is a test of delaying between task runs, not a test of exponential distribution.
+	mockExponentialDelay := 30 * time.Millisecond
+	q.ExponentialBackoffFn = func(failureCount int) time.Duration {
+		return mockExponentialDelay
+	}
+
+	q.Start()
+
+	// Expect TaskHandler returns Success result.
+	g.Eventually(queueStopCh, "5s", "20ms").Should(BeClosed(), "Should handle first task in queue successfully")
+
+	// Expect TaskHandler called 'fails' times.
+	g.Expect(Task.GetFailureCount()).Should(Equal(fails), "task should fail %d times.", fails)
+
+	prev := failureCounts[0]
+	for i := 1; i < len(failureCounts); i++ {
+		cur := failureCounts[i]
+		g.Expect(cur > prev).Should(BeTrue(), "TaskHandler should receive task with growing FailureCount. Got %d after %d", cur, prev)
+	}
+
+	// Expect mean delay is greater than mocked delay.
+	mean, _ := calculateMeanDelay(runsAt)
+	g.Expect(mean).Should(BeNumerically(">", mockExponentialDelay),
+		"mean delay of %d fails should be more than %s, got %s. Check exponential delaying not broken in Start or waitForTask.",
+		fails, mockExponentialDelay.String(), mean.Truncate(100*time.Microsecond).String())
+
+}
+
+func calculateMeanDelay(in []time.Time) (mean time.Duration, deltas []int64) {
+	var sum int64
+
+	// Calculate deltas from timestamps.
+	prev := in[0].UnixNano()
+	for i := 1; i < len(in); i++ {
+		delta := in[i].UnixNano() - prev
+		prev = in[i].UnixNano()
+		deltas = append(deltas, delta)
+		sum += delta
+	}
+	mean = time.Duration(sum / int64(len(deltas)))
+	return
+}
+
+func Test_CancelDelay(t *testing.T) {
+	g := NewWithT(t)
+	// Init and prefill queue.
+	q := NewTasksQueue()
+	q.WithContext(context.TODO())
+	q.WithName("test-queue")
+	// Since we don't want the test to run for too long, we don't
+	// want to use lengthy times.
+	q.WaitLoopCheckInterval = 5 * time.Millisecond // default is 125ms
+	q.DelayOnQueueIsEmpty = 5 * time.Millisecond   // default is 250ms
+	q.DelayOnRepeat = 5 * time.Millisecond         // default is 25ms
+	// Add 'always fail' task.
+	ErrTask := &task.BaseTask{Id: "erroneous"}
+	q.AddFirst(ErrTask)
+
+	HealingTask := &task.BaseTask{Id: "healing"}
+
+	// Set handler to always return fail for task "erroneous", and return success for other tasks.
+	// Catch time between "erroneous" task handling and "healing" task handling.
+	startedAt := time.Now()
+	endedAt := startedAt
+	delayStartsCh := make(chan struct{}, 1)
+	healingDoneCh := make(chan struct{}, 1)
+	q.WithHandler(func(t task.Task) (res TaskResult) {
+		if t.GetId() == ErrTask.GetId() {
+			res.Status = Fail
+			// Close chan after first delay.
+			if t.GetFailureCount() == 1 {
+				res.AfterHandle = func() {
+					close(delayStartsCh)
+				}
+			}
+			return
+		}
+		if t.GetId() == HealingTask.GetId() {
+			endedAt = time.Now()
+			res.AfterHandle = func() {
+				close(healingDoneCh)
+			}
+		}
+		res.Status = Success
+		return
+	})
+
+	// Set exponential backoff to the constant delay just to wait more than DelayOnQueueIsEmpty.
+	// It is a test of delaying between task runs, not a test of exponential distribution.
+	mockExponentialDelay := 150 * time.Millisecond
+	q.ExponentialBackoffFn = func(failureCount int) time.Duration {
+		return mockExponentialDelay
+	}
+
+	// Start handling 'erroneous' task.
+	q.Start()
+
+	// Expect TaskHandler returns Success result.
+	g.Eventually(delayStartsCh, "5s", "20ms").Should(BeClosed(), "Should handle failed task and starts a delay")
+
+	// Add healing task and cancel delay in parallel.
+	go func() {
+		q.AddFirst(HealingTask)
+		q.CancelTaskDelay()
+	}()
+
+	// Expect queue handles 'healing' task.
+	g.Eventually(healingDoneCh, "5s", "20ms").Should(BeClosed(), "Should handle 'healing' task")
+
+	elapsed := endedAt.Sub(startedAt).Truncate(100 * time.Microsecond)
+
+	// Expect elapsed is less than mocked delay.
+	g.Expect(elapsed).Should(BeNumerically(">", mockExponentialDelay),
+		"Should delay after failed task. Got delay of %s, expect more than %s. Check delay for failed task not broken in Start or waitForTask.",
+		elapsed.String(), mockExponentialDelay.String())
+	g.Expect(elapsed).Should(BeNumerically("<", 2*mockExponentialDelay),
+		"Should stop delaying after CancelTaskDelay call. Got delay of %s, expect less than %s. Check cancel delay not broken in Start or waitForTask.",
+		elapsed.String(), (2 * mockExponentialDelay).String())
+}