Merge pull request #160 from cloudflare/repair-exp-backoff

add exponential decay to repair delay

Author: mattalberts

cmd/argot/main.go

@@ -54,6 +54,7 @@ func main() {
 	connlimit := couple.Flag("connection-limit", "profiling bind address").Default("512").Int()
 	repairdelay := couple.Flag("repair-delay", "period between tunnel repair attempts").Default(argotunnel.RepairDelayDefault.String()).Duration()
 	repairjitter := couple.Flag("repair-jitter", "linear jitter as a fraction of repair-delay").Default(strconv.FormatFloat(argotunnel.RepairJitterDefault, 'E', -1, 64)).Float64()
+	repairsteps := couple.Flag("repair-steps", "number of exponential steps used during tunnel repair").Default(strconv.FormatUint(argotunnel.RepairStepsDefault, 10)).Uint()
 	resyncperiod := couple.Flag("resync-period", "period between synchronization attempts").Default(argotunnel.ResyncPeriodDefault.String()).Duration()
 	taglimit := couple.Flag("tag-limit", "number of tags allowed per tunnel").Default(strconv.Itoa(argotunnel.TagLimitDefault)).Int()
 	transportlogenable := couple.Flag("transport-log-enable", "enable transport logging").Bool()
@@ -176,7 +177,7 @@ func main() {
 			}
 
 			argotunnel.EnableMetrics(5 * time.Second)
-			argotunnel.SetRepairBackoff(*repairdelay, *repairjitter)
+			argotunnel.SetRepairBackoff(*repairdelay, *repairjitter, *repairsteps)
 			argotunnel.SetTagLimit(*taglimit)
 			argotunnel.SetVersion(version)
 

hack/tunnel.gosrc

@@ -196,11 +196,11 @@ func ServeTunnel(
 	// Treat panics as recoverable errors
 	defer func() {
 		if r := recover(); r != nil {
-			recoverable = true
-			err = fmt.Errorf("tunnel: %v", r)
 			config.Logger.WithFields(log.Fields{
-				"trace": debug.Stack(),
+				"trace": string(debug.Stack()),
 			}).Errorf("tunnel runtime panic: %v", r)
+			err = fmt.Errorf("tunnel: %v", r)
+			recoverable = true
 		}
 	}()
 

internal/argotunnel/tunnel.go

@@ -6,6 +6,7 @@ import (
 	"crypto/x509"
 	"encoding/pem"
 	"fmt"
+	"math/rand"
 	"net"
 	"net/http"
 	"reflect"
@@ -17,15 +18,16 @@ import (
 	"github.com/cloudflare/cloudflared/origin"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	"github.com/sirupsen/logrus"
-	"k8s.io/apimachinery/pkg/util/rand"
-	"k8s.io/apimachinery/pkg/util/wait"
+	utilrand "k8s.io/apimachinery/pkg/util/rand"
 )
 
 const (
 	// RepairDelayDefault the default base time to wait between repairs
-	RepairDelayDefault = 40 * time.Millisecond
+	RepairDelayDefault = 100 * time.Millisecond
 	// RepairJitterDefault the default linear jitter applied to the wait on repair
-	RepairJitterDefault = 1.0
+	RepairJitterDefault = 0.5
+	// RepairStepsDefault the default exponential steps used during repair
+	RepairStepsDefault = 4
 	// TagLimitDefault the default number of unique tags
 	TagLimitDefault = 32
 
@@ -35,17 +37,20 @@ const (
 var repairBackoff = struct {
 	delay     time.Duration
 	jitter    float64
+	steps     uint
 	setRepair sync.Once
 }{
 	delay:  RepairDelayDefault,
 	jitter: RepairJitterDefault,
+	steps:  RepairStepsDefault,
 }
 
 // SetRepairBackoff configures the repair backoff used by all tunnels
-func SetRepairBackoff(delay time.Duration, jitter float64) {
+func SetRepairBackoff(delay time.Duration, jitter float64, steps uint) {
 	repairBackoff.setRepair.Do(func() {
 		repairBackoff.delay = delay
 		repairBackoff.jitter = jitter
+		repairBackoff.steps = steps
 	})
 }
 
@@ -90,15 +95,16 @@ type tunnelLink interface {
 }
 
 type syncTunnelLink struct {
-	mu     sync.RWMutex
-	rule   tunnelRule
-	cert   []byte
-	opts   tunnelOptions
-	config *origin.TunnelConfig
-	errCh  chan error
-	quitCh chan struct{}
-	stopCh chan struct{}
-	log    *logrus.Logger
+	mu      sync.RWMutex
+	rule    tunnelRule
+	cert    []byte
+	opts    tunnelOptions
+	config  *origin.TunnelConfig
+	errCh   chan error
+	quitCh  chan struct{}
+	stopCh  chan struct{}
+	repiars uint
+	log     *logrus.Logger
 }
 
 func (l *syncTunnelLink) host() string {
@@ -203,7 +209,7 @@ func newLinkTunnelConfig(rule tunnelRule, cert []byte, options tunnelOptions) *o
 		Retries:           options.retries,
 		HeartbeatInterval: options.heartbeatInterval,
 		MaxHeartbeats:     options.heartbeatCount,
-		ClientID:          rand.String(32),
+		ClientID:          utilrand.String(32),
 		BuildInfo:         origin.GetBuildInfo(),
 		ReportedVersion:   versionConfig.version,
 		LBPool:            options.lbPool,
@@ -220,6 +226,8 @@ func newLinkTunnelConfig(rule tunnelRule, cert []byte, options tunnelOptions) *o
 		RunFromTerminal:    false, // bool
 		NoChunkedEncoding:  options.noChunkedEncoding,
 		CompressionQuality: options.compressionQuality,
+		IncidentLookup:     origin.NewIncidentLookup(),
+		CloseConnOnce:      &sync.Once{},
 	}
 }
 
@@ -365,7 +373,7 @@ func repairFunc(l *syncTunnelLink) func() {
 						}).Errorf("link exited with error (%s) '%v', repairing ...", reflect.TypeOf(err), err)
 
 						// linear back-off on runtime error
-						delay := wait.Jitter(repairBackoff.delay, repairBackoff.jitter)
+						delay := repairDelay(ll.repiars, repairBackoff.delay, repairBackoff.jitter, repairBackoff.steps)
 						log.WithFields(logrus.Fields{
 							"origin":   ll.config.OriginUrl,
 							"hostname": ll.rule.host,
@@ -393,12 +401,24 @@ func repairFunc(l *syncTunnelLink) func() {
 						}
 
 						close(ll.stopCh)
-						ll.config.ClientID = rand.String(32)
+						ll.config.ClientID = utilrand.String(32)
 						ll.stopCh = make(chan struct{})
+						ll.repiars++
 						go launchFunc(ll)()
 					}()
 				}
 			}
 		}
 	}
 }
+
+func repairDelay(step uint, delay time.Duration, jitter float64, steps uint) time.Duration {
+	d := delay
+	if steps > 0 {
+		d = (1 << (step % steps)) * delay
+	}
+	if jitter > 0 {
+		d += time.Duration(rand.Float64() * jitter * float64(delay))
+	}
+	return d
+}

internal/argotunnel/tunnel_test.go

@@ -318,32 +318,89 @@ func TestParseTags(t *testing.T) {
 func TestSetRepairBackoff(t *testing.T) {
 	repairDelay := repairBackoff.delay
 	repairJitter := repairBackoff.jitter
+	repairSteps := repairBackoff.steps
 	repairs := []struct {
 		delay  time.Duration
 		jitter float64
+		steps  uint
 	}{
 		{
 			delay:  1 * time.Millisecond,
-			jitter: 0.5,
+			jitter: 0.25,
+			steps:  1,
 		},
 		{
 			delay:  10 * time.Millisecond,
-			jitter: .25,
+			jitter: .125,
+			steps:  2,
 		},
 		{
 			delay:  100 * time.Millisecond,
-			jitter: .125,
+			jitter: .0625,
+			steps:  3,
 		},
 	}
 
 	for _, r := range repairs {
-		SetRepairBackoff(r.delay, r.jitter)
+		SetRepairBackoff(r.delay, r.jitter, r.steps)
 	}
 
 	assert.Equalf(t, repairs[0].delay, repairBackoff.delay, "test repair delay matches first set")
 	assert.Equalf(t, repairs[0].jitter, repairBackoff.jitter, "test repair jitter matches first set")
+	assert.Equalf(t, repairs[0].steps, repairBackoff.steps, "test repair steps matches first set")
 	assert.NotEqualf(t, repairBackoff.delay, repairDelay, "test repair delay does not match default")
 	assert.NotEqualf(t, repairBackoff.jitter, repairJitter, "test repair jitter does not match default")
+	assert.NotEqualf(t, repairBackoff.steps, repairSteps, "test repair steps does not match default")
+}
+
+func TestRepairDelay(t *testing.T) {
+	t.Parallel()
+	for name, test := range map[string]struct {
+		step   uint
+		delay  time.Duration
+		jitter float64
+		steps  uint
+		out    time.Duration
+	}{
+		"step-0-no-jitter": {
+			step:   0,
+			delay:  10 * time.Millisecond,
+			jitter: 0.0,
+			steps:  4,
+			out:    10 * time.Millisecond,
+		},
+		"step-0-no-jitter-no-steps": {
+			step:   0,
+			delay:  10 * time.Millisecond,
+			jitter: 0.0,
+			steps:  0,
+			out:    10 * time.Millisecond,
+		},
+		"step-1-no-jitter": {
+			step:   1,
+			delay:  10 * time.Millisecond,
+			jitter: 0.0,
+			steps:  4,
+			out:    20 * time.Millisecond,
+		},
+		"step-2-no-jitter": {
+			step:   2,
+			delay:  10 * time.Millisecond,
+			jitter: 0.0,
+			steps:  4,
+			out:    40 * time.Millisecond,
+		},
+		"step-4-no-jitter": {
+			step:   4,
+			delay:  10 * time.Millisecond,
+			jitter: 0.0,
+			steps:  4,
+			out:    10 * time.Millisecond,
+		},
+	} {
+		out := repairDelay(test.step, test.delay, test.jitter, test.steps)
+		assert.Equalf(t, test.out, out, "test '%s' value mismatch", name)
+	}
 }
 
 func TestSetTagLimit(t *testing.T) {