Merge pull request #17 from utilitywarehouse/retry-backoff

Add a backoff retry mechanism for starting runners

Author: ffilippopoulos

Dockerfile

@@ -5,7 +5,7 @@ ENV CGO_ENABLED=0
 RUN \
   apk --no-cache add git upx \
   && go get -t ./... \
-  && go test -v \
+  && go test -v ./... \
   && go build -ldflags='-s -w' -o /semaphore-wireguard . \
   && upx /semaphore-wireguard
 

backoff/backoff.go

@@ -0,0 +1,100 @@
+// Package backoff includes this backoff function copied from: https://github.com/jpillora/backoff/blob/d80867952dff4e2fbfb4280ded4ff94d67790457/backoff.go
+package backoff
+
+import (
+	"math"
+	"math/rand"
+	"sync/atomic"
+	"time"
+)
+
+// Backoff is a time.Duration counter, starting at Min. After every call to
+// the Duration method the current timing is multiplied by Factor, but it
+// never exceeds Max.
+//
+// Backoff is not generally concurrent-safe, but the ForAttempt method can
+// be used concurrently.
+type Backoff struct {
+	attempt uint64
+	// Factor is the multiplying factor for each increment step
+	Factor float64
+	// Jitter eases contention by randomizing backoff steps
+	Jitter bool
+	// Min and Max are the minimum and maximum values of the counter
+	Min, Max time.Duration
+}
+
+// Duration returns the duration for the current attempt before incrementing
+// the attempt counter. See ForAttempt.
+func (b *Backoff) Duration() time.Duration {
+	d := b.ForAttempt(float64(atomic.AddUint64(&b.attempt, 1) - 1))
+	return d
+}
+
+const maxInt64 = float64(math.MaxInt64 - 512)
+
+// ForAttempt returns the duration for a specific attempt. This is useful if
+// you have a large number of independent Backoffs, but don't want use
+// unnecessary memory storing the Backoff parameters per Backoff. The first
+// attempt should be 0.
+//
+// ForAttempt is concurrent-safe.
+func (b *Backoff) ForAttempt(attempt float64) time.Duration {
+	// Zero-values are nonsensical, so we use
+	// them to apply defaults
+	min := b.Min
+	if min <= 0 {
+		min = 100 * time.Millisecond
+	}
+	max := b.Max
+	if max <= 0 {
+		max = 10 * time.Second
+	}
+	if min >= max {
+		// short-circuit
+		return max
+	}
+	factor := b.Factor
+	if factor <= 0 {
+		factor = 2
+	}
+	//calculate this duration
+	minf := float64(min)
+	durf := minf * math.Pow(factor, attempt)
+	if b.Jitter {
+		durf = rand.Float64()*(durf-minf) + minf
+	}
+	//ensure float64 wont overflow int64
+	if durf > maxInt64 {
+		return max
+	}
+	dur := time.Duration(durf)
+	//keep within bounds
+	if dur < min {
+		return min
+	}
+	if dur > max {
+		return max
+	}
+	return dur
+}
+
+// Reset restarts the current attempt counter at zero.
+func (b *Backoff) Reset() {
+	atomic.StoreUint64(&b.attempt, 0)
+}
+
+// Attempt returns the current attempt counter value.
+func (b *Backoff) Attempt() float64 {
+	return float64(atomic.LoadUint64(&b.attempt))
+}
+
+// Copy returns a backoff with equals constraints as the original
+func (b *Backoff) Copy() *Backoff {
+	return &Backoff{
+		Factor: b.Factor,
+		Jitter: b.Jitter,
+		Min:    b.Min,
+		Max:    b.Max,
+	}
+}

backoff/retry.go

@@ -0,0 +1,44 @@
+package backoff
+
+import (
+	"time"
+
+	"github.com/utilitywarehouse/semaphore-wireguard/log"
+)
+
+type operation func() error
+
+const (
+	defaultBackoffJitter = true
+	defaultBackoffMin    = 2 * time.Second
+	defaultBackoffMax    = 1 * time.Minute
+)
+
+// Retry will use the default backoff values to retry the passed operation
+func Retry(op operation, description string) {
+	b := &Backoff{
+		Jitter: defaultBackoffJitter,
+		Min:    defaultBackoffMin,
+		Max:    defaultBackoffMax,
+	}
+	RetryWithBackoff(op, b, description)
+}
+
+// RetryWithBackoff will retry the passed function (operation) using the given
+// backoff
+func RetryWithBackoff(op operation, b *Backoff, description string) {
+	b.Reset()
+	for {
+		err := op()
+		if err == nil {
+			return
+		}
+		d := b.Duration()
+		log.Logger.Error("Retry failed",
+			"description", description,
+			"error", err,
+			"backoff", d,
+		)
+		time.Sleep(d)
+	}
+}

backoff/retry_test.go

@@ -0,0 +1,39 @@
+package backoff
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/utilitywarehouse/semaphore-wireguard/log"
+)
+
+var testFuncCallCounter int
+var successThreshold int
+
+func testFunc() error {
+	testFuncCallCounter++
+
+	if testFuncCallCounter >= successThreshold {
+		return nil
+	}
+	return errors.New("error")
+
+}
+
+func TestRetryWithBackoff(t *testing.T) {
+	log.InitLogger("retry-test", "info")
+	b := &Backoff{
+		Jitter: false,
+		Min:    10 * time.Millisecond,
+		Max:    1 * time.Second,
+	}
+	successThreshold = 3
+
+	// Retrying testFunc should fail 2 times before hitting the success
+	// threshold
+	RetryWithBackoff(testFunc, b, "test func")
+	assert.Equal(t, testFuncCallCounter, 3)            // should be 3 after 2 consecutive fails
+	assert.Equal(t, b.Duration(), 40*time.Millisecond) // should be 40 millisec after failing for 10 and 20 and without a jitter
+}

config.go

@@ -11,15 +11,18 @@ const (
 	defaultWGListenPort = 51820
 )
 
+// Duration is a helper to unmarshal time.Duration from json
 // https://stackoverflow.com/questions/48050945/how-to-unmarshal-json-into-durations/54571600#54571600
 type Duration struct {
 	time.Duration
 }
 
+// MarshalJSON calls json Marshall on Duration
 func (d Duration) MarshalJSON() ([]byte, error) {
 	return json.Marshal(d.String())
 }
 
+// UnmarshalJSON provides handling of time.Duration when unmarshalling
 func (d *Duration) UnmarshalJSON(b []byte) error {
 	var v interface{}
 	if err := json.Unmarshal(b, &v); err != nil {
@@ -58,6 +61,7 @@ type remoteClusterConfig struct {
 	ResyncPeriod      Duration `json:"resyncPeriod"`
 }
 
+// Config holds the application configuration
 type Config struct {
 	Local   localClusterConfig     `json:"local"`
 	Remotes []*remoteClusterConfig `json:"remotes"`

main.go

@@ -12,6 +12,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"golang.zx2c4.com/wireguard/wgctrl"
 
+	"github.com/utilitywarehouse/semaphore-wireguard/backoff"
 	"github.com/utilitywarehouse/semaphore-wireguard/kube"
 	"github.com/utilitywarehouse/semaphore-wireguard/log"
 	"k8s.io/client-go/kubernetes"
@@ -89,9 +90,7 @@ func main() {
 		wgDeviceNames = append(wgDeviceNames, wgDeviceName)
 		runners = append(runners, r)
 		go func() {
-			if err := r.Run(); err != nil {
-				log.Logger.Error("Failed to start runner", "err", err)
-			}
+			backoff.Retry(r.Run, "start runner")
 		}()
 	}
 

wireguard/wireguard_test.go

@@ -11,28 +11,28 @@ var (
 
 func TestNewPeerConfig(t *testing.T) {
 	var err error
-	_, err = newPeerConfig("", "", "", nil)
+	_, err = NewPeerConfig("", "", "", nil)
 	if err == nil {
-		t.Errorf("newPeerConfig: empty publicKey should generate an error")
+		t.Errorf("NewPeerConfig: empty publicKey should generate an error")
 	}
-	_, err = newPeerConfig("foobar", "", "", nil)
+	_, err = NewPeerConfig("foobar", "", "", nil)
 	if err == nil {
-		t.Errorf("newPeerConfig: invalid publicKey should generate an error")
+		t.Errorf("NewPeerConfig: invalid publicKey should generate an error")
 	}
-	_, err = newPeerConfig(validPublicKey, "", "", []string{""})
+	_, err = NewPeerConfig(validPublicKey, "", "", []string{""})
 	if err == nil {
-		t.Errorf("newPeerConfig: invalid allowedIPs should generate an error")
+		t.Errorf("NewPeerConfig: invalid allowedIPs should generate an error")
 	}
-	_, err = newPeerConfig(validPublicKey, "foo", "", validAllowedIPs)
+	_, err = NewPeerConfig(validPublicKey, "foo", "", validAllowedIPs)
 	if err == nil {
-		t.Errorf("newPeerConfig: invalid presharedKey should generate an error")
+		t.Errorf("NewPeerConfig: invalid presharedKey should generate an error")
 	}
-	_, err = newPeerConfig(validPublicKey, validPublicKey, "foo", validAllowedIPs)
+	_, err = NewPeerConfig(validPublicKey, validPublicKey, "foo", validAllowedIPs)
 	if err == nil {
-		t.Errorf("newPeerConfig: invalid endpoint should generate an error")
+		t.Errorf("NewPeerConfig: invalid endpoint should generate an error")
 	}
-	_, err = newPeerConfig(validPublicKey, validPublicKey, "1.1.1.1:1111", validAllowedIPs)
+	_, err = NewPeerConfig(validPublicKey, validPublicKey, "1.1.1.1:1111", validAllowedIPs)
 	if err != nil {
-		t.Errorf("newPeerConfig: unexpected error: %v", err)
+		t.Errorf("NewPeerConfig: unexpected error: %v", err)
 	}
 }