etcdserver: adjust election timeout on restart

etcdserver/membership/cluster.go

@@ -52,6 +52,9 @@ type RaftCluster struct {
 	// removed contains the ids of removed members in the cluster.
 	// removed id cannot be reused.
 	removed map[types.ID]bool
+
+	initAddNotifyOnce *sync.Once
+	initAddNotifyCh   chan struct{}
 }
 
 func NewClusterFromURLsMap(token string, urlsmap types.URLsMap) (*RaftCluster, error) {
@@ -81,9 +84,11 @@ func NewClusterFromMembers(token string, id types.ID, membs []*Member) *RaftClus
 
 func NewCluster(token string) *RaftCluster {
 	return &RaftCluster{
-		token:   token,
-		members: make(map[types.ID]*Member),
-		removed: make(map[types.ID]bool),
+		token:             token,
+		members:           make(map[types.ID]*Member),
+		removed:           make(map[types.ID]bool),
+		initAddNotifyOnce: new(sync.Once),
+		initAddNotifyCh:   make(chan struct{}),
 	}
 }
 
@@ -295,9 +300,16 @@ func (c *RaftCluster) AddMember(m *Member) {
 
 	c.members[m.ID] = m
 
+	if c.initAddNotifyOnce != nil {
+		c.initAddNotifyOnce.Do(func() { close(c.initAddNotifyCh) })
+	}
 	plog.Infof("added member %s %v to cluster %s", m.ID, m.PeerURLs, c.id)
 }
 
+// InitialAddNotify returns a channel that closes when
+// the first member is added to the cluster
+func (c *RaftCluster) InitialAddNotify() <-chan struct{} { return c.initAddNotifyCh }
+
 // RemoveMember removes a member from the store.
 // The given id MUST exist, or the function panics.
 func (c *RaftCluster) RemoveMember(id types.ID) {

etcdserver/raft.go

@@ -97,6 +97,7 @@ type raftNode struct {
 	term         uint64
 	lead         uint64
 
+	tickMu *sync.Mutex
 	raftNodeConfig
 
 	// a chan to send/receive snapshot
@@ -133,6 +134,7 @@ type raftNodeConfig struct {
 
 func newRaftNode(cfg raftNodeConfig) *raftNode {
 	r := &raftNode{
+		tickMu:         new(sync.Mutex),
 		raftNodeConfig: cfg,
 		// set up contention detectors for raft heartbeat message.
 		// expect to send a heartbeat within 2 heartbeat intervals.
@@ -151,6 +153,14 @@ func newRaftNode(cfg raftNodeConfig) *raftNode {
 	return r
 }
 
+// package raft does not have locks
+// prevent racey tick operation from advancing election ticks
+func (r *raftNode) tick() {
+	r.tickMu.Lock()
+	r.raftNodeConfig.Node.Tick()
+	r.tickMu.Unlock()
+}
+
 // start prepares and starts raftNode in a new goroutine. It is no longer safe
 // to modify the fields after it has been started.
 func (r *raftNode) start(rh *raftReadyHandler) {
@@ -163,7 +173,7 @@ func (r *raftNode) start(rh *raftReadyHandler) {
 		for {
 			select {
 			case <-r.ticker.C:
-				r.Tick()
+				r.tick()
 			case rd := <-r.Ready():
 				if rd.SoftState != nil {
 					newLeader := rd.SoftState.Lead != raft.None && atomic.LoadUint64(&r.lead) != rd.SoftState.Lead
@@ -370,16 +380,6 @@ func (r *raftNode) resumeSending() {
 	p.Resume()
 }
 
-// advanceTicksForElection advances ticks to the node for fast election.
-// This reduces the time to wait for first leader election if bootstrapping the whole
-// cluster, while leaving at least 1 heartbeat for possible existing leader
-// to contact it.
-func advanceTicksForElection(n raft.Node, electionTicks int) {
-	for i := 0; i < electionTicks-1; i++ {
-		n.Tick()
-	}
-}
-
 func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id types.ID, n raft.Node, s *raft.MemoryStorage, w *wal.WAL) {
 	var err error
 	member := cl.MemberByName(cfg.Name)
@@ -418,7 +418,6 @@ func startNode(cfg ServerConfig, cl *membership.RaftCluster, ids []types.ID) (id
 	raftStatusMu.Lock()
 	raftStatus = n.Status
 	raftStatusMu.Unlock()
-	advanceTicksForElection(n, c.ElectionTick)
 	return id, n, s, w
 }
 
@@ -453,7 +452,6 @@ func restartNode(cfg ServerConfig, snapshot *raftpb.Snapshot) (types.ID, *member
 	raftStatusMu.Lock()
 	raftStatus = n.Status
 	raftStatusMu.Unlock()
-	advanceTicksForElection(n, c.ElectionTick)
 	return id, cl, n, s, w
 }
 

etcdserver/server.go

@@ -179,6 +179,12 @@ type EtcdServer struct {
 	consistIndex consistentIndex // must use atomic operations to access; keep 64-bit aligned.
 	r            raftNode        // uses 64-bit atomics; keep 64-bit aligned.
 
+	// advanceRaftTicks advances ticks of Raft node.
+	// This can be used for fast-forwarding election
+	// ticks in multi data-center deployments, thus
+	// speeding up election process.
+	advanceRaftTicks func(ticks int)
+
 	readych chan struct{}
 	Cfg     ServerConfig
 
@@ -445,6 +451,12 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) {
 		hostWhitelist: cfg.HostWhitelist,
 	}
 
+	srv.advanceRaftTicks = func(ticks int) {
+		for i := 0; i < ticks; i++ {
+			srv.r.tick()
+		}
+	}
+
 	srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
 
 	srv.be = be
@@ -527,6 +539,62 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) {
 	}
 	srv.r.transport = tr
 
+	// fresh start
+	if !haveWAL {
+		ticks := cfg.ElectionTicks - 1
+		plog.Infof("%s started anew; fast-forwarding %d ticks (election ticks %d) with %d found member(s)", srv.ID(), ticks, cfg.ElectionTicks, len(cl.Members()))
+		srv.advanceRaftTicks(ticks)
+		return srv, nil
+	}
+
+	srv.goAttach(func() {
+		select {
+		case <-cl.InitialAddNotify():
+			singleNode := len(cl.Members()) == 1
+			if !singleNode {
+				break // multi-node
+			}
+
+			// more member-add commands may be applied
+			// then not single-node cluster
+			select {
+			case <-time.After(srv.Cfg.electionTimeout()):
+				singleNode = len(cl.Members()) == 1
+			case <-tr.InitialPeerNotify():
+				singleNode = false
+			}
+
+			// restarted single-node
+			if singleNode {
+				if !srv.isLeader() { // and leader has not been elected
+					ticks := cfg.ElectionTicks - 1
+					plog.Infof("%s as 1-node cluster; fast-forwarding %d ticks (election ticks %d)", srv.ID(), ticks, cfg.ElectionTicks)
+					srv.advanceRaftTicks(ticks)
+				} else {
+					plog.Infof("%s started as leader to 1-node cluster", srv.ID())
+				}
+				return
+			}
+
+		case <-time.After(rafthttp.ConnReadTimeout):
+			// slow raft config change apply
+			plog.Infof("%s waited %s for member add apply but timed out", srv.ID(), rafthttp.ConnReadTimeout)
+			return
+		}
+
+		// multi-node, wait for peer connection reports
+		select {
+		case <-tr.InitialPeerNotify():
+			// adjust ticks in case slow leader message receive
+			ticks := cfg.ElectionTicks - 3
+			plog.Infof("%s initialzed peer connection; fast-forwarding %d ticks (election ticks %d) with %d found member(s)", srv.ID(), ticks, cfg.ElectionTicks, len(cl.Members()))
+			srv.advanceRaftTicks(ticks)
+
+		case <-time.After(rafthttp.ConnReadTimeout):
+			// connection failed, or no active peers
+			plog.Infof("%s waited %s but no active peer found (or restarted 1-node cluster)", srv.ID(), rafthttp.ConnReadTimeout)
+		}
+	})
 	return srv, nil
 }
 

etcdserver/util_test.go

@@ -86,4 +86,5 @@ func (s *nopTransporterWithActiveTime) ActiveSince(id types.ID) time.Time   { re
 func (s *nopTransporterWithActiveTime) Stop()                               {}
 func (s *nopTransporterWithActiveTime) Pause()                              {}
 func (s *nopTransporterWithActiveTime) Resume()                             {}
+func (s *nopTransporterWithActiveTime) InitialPeerNotify() <-chan struct{}  { return nil }
 func (s *nopTransporterWithActiveTime) reset(am map[types.ID]time.Time)     { s.activeMap = am }

rafthttp/peer.go

@@ -122,7 +122,7 @@ func startPeer(transport *Transport, urls types.URLs, peerID types.ID, fs *stats
 	plog.Infof("starting peer %s...", peerID)
 	defer plog.Infof("started peer %s", peerID)
 
-	status := newPeerStatus(peerID)
+	status := newPeerStatus(peerID, transport.initPeerNotifyOnce, transport.initPeerNotifyCh)
 	picker := newURLPicker(urls)
 	errorc := transport.ErrorC
 	r := transport.Raft

rafthttp/peer_status.go

@@ -32,11 +32,16 @@ type peerStatus struct {
 	mu     sync.Mutex // protect variables below
 	active bool
 	since  time.Time
+
+	once   *sync.Once
+	notify chan struct{}
 }
 
-func newPeerStatus(id types.ID) *peerStatus {
+func newPeerStatus(id types.ID, once *sync.Once, notify chan struct{}) *peerStatus {
 	return &peerStatus{
-		id: id,
+		id:     id,
+		once:   once,
+		notify: notify,
 	}
 }
 
@@ -47,6 +52,13 @@ func (s *peerStatus) activate() {
 		plog.Infof("peer %s became active", s.id)
 		s.active = true
 		s.since = time.Now()
+
+		if s.once != nil {
+			s.once.Do(func() {
+				plog.Infof("notifying of active peer %q", s.id)
+				close(s.notify)
+			})
+		}
 	}
 }
 

rafthttp/peer_test.go

@@ -15,8 +15,11 @@
 package rafthttp
 
 import (
+	"sync"
 	"testing"
+	"time"
 
+	"github.com/coreos/etcd/pkg/testutil"
 	"github.com/coreos/etcd/raft/raftpb"
 )
 
@@ -85,3 +88,24 @@ func TestPeerPick(t *testing.T) {
 		}
 	}
 }
+
+func TestInitialPeerNotify(t *testing.T) {
+	tr := &roundTripperRecorder{rec: testutil.NewRecorderStream()}
+	picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
+	tp := &Transport{
+		pipelineRt:         tr,
+		initPeerNotifyOnce: &sync.Once{},
+		initPeerNotifyCh:   make(chan struct{}),
+	}
+	p := startTestPipeline(tp, picker)
+	defer p.stop()
+
+	p.msgc <- raftpb.Message{Type: raftpb.MsgApp}
+	tr.rec.Wait(1)
+
+	select {
+	case <-tp.InitialPeerNotify():
+	case <-time.After(3 * time.Second):
+		t.Fatal("took too long to receive initial peer notify")
+	}
+}

rafthttp/pipeline_test.go

@@ -301,7 +301,7 @@ func startTestPipeline(tr *Transport, picker *urlPicker) *pipeline {
 		peerID:        types.ID(1),
 		tr:            tr,
 		picker:        picker,
-		status:        newPeerStatus(types.ID(1)),
+		status:        newPeerStatus(types.ID(1), tr.initPeerNotifyOnce, tr.initPeerNotifyCh),
 		raft:          &fakeRaft{},
 		followerStats: &stats.FollowerStats{},
 		errorc:        make(chan error, 1),

rafthttp/remote.go

@@ -27,7 +27,7 @@ type remote struct {
 
 func startRemote(tr *Transport, urls types.URLs, id types.ID) *remote {
 	picker := newURLPicker(urls)
-	status := newPeerStatus(id)
+	status := newPeerStatus(id, tr.initPeerNotifyOnce, tr.initPeerNotifyCh)
 	pipeline := &pipeline{
 		peerID: id,
 		tr:     tr,

rafthttp/snapshot_test.go

@@ -107,7 +107,7 @@ func testSnapshotSend(t *testing.T, sm *raftsnap.Message) (bool, []os.FileInfo)
 	defer srv.Close()
 
 	picker := mustNewURLPicker(t, []string{srv.URL})
-	snapsend := newSnapshotSender(tr, picker, types.ID(1), newPeerStatus(types.ID(1)))
+	snapsend := newSnapshotSender(tr, picker, types.ID(1), newPeerStatus(types.ID(1), nil, nil))
 	defer snapsend.stop()
 
 	snapsend.send(*sm)

rafthttp/stream_test.go

@@ -40,7 +40,7 @@ import (
 // to streamWriter. After that, streamWriter can use it to send messages
 // continuously, and closes it when stopped.
 func TestStreamWriterAttachOutgoingConn(t *testing.T) {
-	sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
+	sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1), nil, nil), &stats.FollowerStats{}, &fakeRaft{})
 	// the expected initial state of streamWriter is not working
 	if _, ok := sw.writec(); ok {
 		t.Errorf("initial working status = %v, want false", ok)
@@ -92,7 +92,7 @@ func TestStreamWriterAttachOutgoingConn(t *testing.T) {
 // TestStreamWriterAttachBadOutgoingConn tests that streamWriter with bad
 // outgoingConn will close the outgoingConn and fall back to non-working status.
 func TestStreamWriterAttachBadOutgoingConn(t *testing.T) {
-	sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
+	sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1), nil, nil), &stats.FollowerStats{}, &fakeRaft{})
 	defer sw.stop()
 	wfc := newFakeWriteFlushCloser(errors.New("blah"))
 	sw.attach(&outgoingConn{t: streamTypeMessage, Writer: wfc, Flusher: wfc, Closer: wfc})
@@ -196,7 +196,7 @@ func TestStreamReaderStopOnDial(t *testing.T) {
 		picker: mustNewURLPicker(t, []string{"http://localhost:2380"}),
 		errorc: make(chan error, 1),
 		typ:    streamTypeMessage,
-		status: newPeerStatus(types.ID(2)),
+		status: newPeerStatus(types.ID(2), nil, nil),
 		rl:     rate.NewLimiter(rate.Every(100*time.Millisecond), 1),
 	}
 	tr.onResp = func() {
@@ -303,7 +303,7 @@ func TestStream(t *testing.T) {
 		srv := httptest.NewServer(h)
 		defer srv.Close()
 
-		sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
+		sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1), nil, nil), &stats.FollowerStats{}, &fakeRaft{})
 		defer sw.stop()
 		h.sw = sw
 
@@ -315,7 +315,7 @@ func TestStream(t *testing.T) {
 			typ:    tt.t,
 			tr:     tr,
 			picker: picker,
-			status: newPeerStatus(types.ID(2)),
+			status: newPeerStatus(types.ID(2), nil, nil),
 			recvc:  recvc,
 			propc:  propc,
 			rl:     rate.NewLimiter(rate.Every(100*time.Millisecond), 1),