dispatcher: Give a couple of minutes for nodes to recover on re-election

manager/dispatcher/dispatcher.go

@@ -31,6 +31,7 @@ const (
 	defaultHeartBeatEpsilon      = 500 * time.Millisecond
 	defaultGracePeriodMultiplier = 3
 	defaultRateLimitPeriod       = 8 * time.Second
+	defaultStartupGracePeriod    = 2 * time.Minute
 
 	// maxBatchItems is the threshold of queued writes that should
 	// trigger an actual transaction to commit them to the shared store.
@@ -61,9 +62,10 @@ var (
 // DefautConfig.
 type Config struct {
 	// Addr configures the address the dispatcher reports to agents.
-	Addr             string
-	HeartbeatPeriod  time.Duration
-	HeartbeatEpsilon time.Duration
+	Addr               string
+	HeartbeatPeriod    time.Duration
+	HeartbeatEpsilon   time.Duration
+	StartupGracePeriod time.Duration
 	// RateLimitPeriod specifies how often node with same ID can try to register
 	// new session.
 	RateLimitPeriod       time.Duration
@@ -75,6 +77,7 @@ func DefaultConfig() *Config {
 	return &Config{
 		HeartbeatPeriod:       DefaultHeartBeatPeriod,
 		HeartbeatEpsilon:      defaultHeartBeatEpsilon,
+		StartupGracePeriod:    defaultStartupGracePeriod,
 		RateLimitPeriod:       defaultRateLimitPeriod,
 		GracePeriodMultiplier: defaultGracePeriodMultiplier,
 	}
@@ -122,7 +125,7 @@ func (b weightedPeerByNodeID) Swap(i, j int) { b[i], b[j] = b[j], b[i] }
 func New(cluster Cluster, c *Config) *Dispatcher {
 	return &Dispatcher{
 		addr:                      c.Addr,
-		nodes:                     newNodeStore(c.HeartbeatPeriod, c.HeartbeatEpsilon, c.GracePeriodMultiplier, c.RateLimitPeriod),
+		nodes:                     newNodeStore(c.HeartbeatPeriod, c.HeartbeatEpsilon, c.GracePeriodMultiplier, c.StartupGracePeriod, c.RateLimitPeriod),
 		store:                     cluster.MemoryStore(),
 		cluster:                   cluster,
 		mgrQueue:                  watch.NewQueue(16),

manager/dispatcher/dispatcher_test.go

@@ -273,6 +273,7 @@ func TestHeartbeatTimeout(t *testing.T) {
 	cfg := DefaultConfig()
 	cfg.HeartbeatPeriod = 100 * time.Millisecond
 	cfg.HeartbeatEpsilon = 0
+	cfg.StartupGracePeriod = 100 * time.Millisecond
 	gd, err := startDispatcher(cfg)
 	assert.NoError(t, err)
 	defer gd.Close()
@@ -737,6 +738,7 @@ func TestNodesCount(t *testing.T) {
 	cfg := DefaultConfig()
 	cfg.HeartbeatPeriod = 100 * time.Millisecond
 	cfg.HeartbeatEpsilon = 0
+	cfg.StartupGracePeriod = 100 * time.Millisecond
 	gd, err := startDispatcher(cfg)
 	assert.NoError(t, err)
 	defer gd.Close()

manager/dispatcher/nodes.go

@@ -12,7 +12,9 @@ import (
 	"github.com/docker/swarmkit/manager/dispatcher/heartbeat"
 )
 
-const rateLimitCount = 3
+const (
+	rateLimitCount = 3
+)
 
 type registeredNode struct {
 	SessionID  string
@@ -45,16 +47,18 @@ func (rn *registeredNode) checkSessionID(sessionID string) error {
 type nodeStore struct {
 	periodChooser         *periodChooser
 	gracePeriodMultiplier time.Duration
+	startupGracePeriod    time.Duration
 	rateLimitPeriod       time.Duration
 	nodes                 map[string]*registeredNode
 	mu                    sync.RWMutex
 }
 
-func newNodeStore(hbPeriod, hbEpsilon time.Duration, graceMultiplier int, rateLimitPeriod time.Duration) *nodeStore {
+func newNodeStore(hbPeriod, hbEpsilon time.Duration, graceMultiplier int, startupGracePeriod, rateLimitPeriod time.Duration) *nodeStore {
 	return &nodeStore{
 		nodes:                 make(map[string]*registeredNode),
 		periodChooser:         newPeriodChooser(hbPeriod, hbEpsilon),
 		gracePeriodMultiplier: time.Duration(graceMultiplier),
+		startupGracePeriod:    startupGracePeriod,
 		rateLimitPeriod:       rateLimitPeriod,
 	}
 }
@@ -79,7 +83,8 @@ func (s *nodeStore) AddUnknown(n *api.Node, expireFunc func()) error {
 		Node: n,
 	}
 	s.nodes[n.ID] = rn
-	rn.Heartbeat = heartbeat.New(s.periodChooser.Choose()*s.gracePeriodMultiplier, expireFunc)
+	hb := s.periodChooser.Choose()*s.gracePeriodMultiplier + s.startupGracePeriod
+	rn.Heartbeat = heartbeat.New(hb, expireFunc)
 	return nil
 }