1. Refactor:change TigLabs raft Authors to tiglabs raft Authors.

2. Refactor: add tiglabs modify if the source file is etcd authors
3. Fix: when raft member become leader,must apply from appliyID to commitID
4. Refactor: add raft heartbear check, if no leader,then warnint
5. Fix: truncate raft corrupt data
6. Refactor: raft heartbeat maybe block when peers host down

Signed-off-by: awzhgw <guowl18702995996@gmail.com>

Author: awzhgw

config.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

errors.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

future.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The TigLabs raft Authors.
+// Copyright 2018 The tiglabs raft Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

logger/logger.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

pool.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The TigLabs raft Authors.
+// Copyright 2018 The tiglabs raft Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

proto/codec.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The TigLabs raft Authors.
+// Copyright 2018 The tiglabs raft Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

proto/pool.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The TigLabs raft Authors.
+// Copyright 2018 The tiglabs raft Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

proto/proto.go

@@ -1,4 +1,4 @@
-// Copyright 2018 The TigLabs raft Authors.
+// Copyright 2018 The tiglabs raft Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -115,6 +115,12 @@ type Message struct {
 	Snapshot     Snapshot // No need for codec
 }
 
+func (m *Message) ToString() (mesg string) {
+	return fmt.Sprintf("Mesg:[%v] type(%v) ForceVote(%v) Reject(%v) RejectIndex(%v) "+
+		"From(%v) To(%v) Term(%v) LogTrem(%v) Index(%v) Commit(%v)", m.ID, m.Type.String(), m.ForceVote,
+		m.Reject, m.RejectIndex, m.From, m.To, m.Term, m.LogTerm, m.Index, m.Commit)
+}
+
 type ConfChange struct {
 	Type    ConfChangeType
 	Peer    Peer

raft.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,6 +20,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 	"unsafe"
 
 	"github.com/tiglabs/raft/logger"
@@ -58,6 +60,11 @@ type peerState struct {
 	mu    sync.RWMutex
 }
 
+type monitorStatus struct {
+	conErrCount    uint8
+	replicasErrCnt map[uint64]uint8
+}
+
 func (s *peerState) change(c *proto.ConfChange) {
 	s.mu.Lock()
 	switch c.Type {
@@ -102,6 +109,7 @@ type raft struct {
 	peerState         peerState
 	pending           map[uint64]*Future
 	snapping          map[uint64]*snapshotStatus
+	mStatus           *monitorStatus
 	propc             chan *proposal
 	applyc            chan *apply
 	recvc             chan *proto.Message
@@ -130,10 +138,15 @@ func newRaft(config *Config, raftConfig *RaftConfig) (*raft, error) {
 		return nil, err
 	}
 
+	mStatus := &monitorStatus{
+		conErrCount:    0,
+		replicasErrCnt: make(map[uint64]uint8),
+	}
 	raft := &raft{
 		raftFsm:       r,
 		config:        config,
 		raftConfig:    raftConfig,
+		mStatus:       mStatus,
 		pending:       make(map[uint64]*Future),
 		snapping:      make(map[uint64]*snapshotStatus),
 		recvc:         make(chan *proto.Message, config.ReqBufferSize),
@@ -155,6 +168,7 @@ func newRaft(config *Config, raftConfig *RaftConfig) (*raft, error) {
 
 	util.RunWorker(raft.runApply, raft.handlePanic)
 	util.RunWorker(raft.run, raft.handlePanic)
+	util.RunWorker(raft.monitor, raft.handlePanic)
 	return raft, nil
 }
 
@@ -388,6 +402,51 @@ func (s *raft) run() {
 	}
 }
 
+func (s *raft) monitor() {
+	statusTicker := time.NewTicker(5 * time.Second)
+	leaderTicker := time.NewTicker(1 * time.Minute)
+	for {
+		select {
+		case <-s.stopc:
+			statusTicker.Stop()
+			return
+
+		case <-statusTicker.C:
+			if s.raftFsm.leader == NoLeader || s.raftFsm.state == stateCandidate {
+				s.mStatus.conErrCount++
+			} else {
+				s.mStatus.conErrCount = 0
+			}
+			if s.mStatus.conErrCount > 5 {
+				errMsg := fmt.Sprintf("raft status not health partitionID[%d]_nodeID[%d]_leader[%v]_state[%v]_replicas[%v]",
+					s.raftFsm.id, s.raftFsm.config.NodeID, s.raftFsm.leader, s.raftFsm.state, s.raftFsm.peers())
+				logger.Error(errMsg)
+
+				s.mStatus.conErrCount = 0
+			}
+		case <-leaderTicker.C:
+			if s.raftFsm.state == stateLeader {
+				for id, p := range s.raftFsm.replicas {
+					if id == s.raftFsm.config.NodeID {
+						continue
+					}
+					if p.active == false {
+						s.mStatus.replicasErrCnt[id]++
+					} else {
+						s.mStatus.replicasErrCnt[id] = 0
+					}
+					if s.mStatus.replicasErrCnt[id] > 5 {
+						errMsg := fmt.Sprintf("raft partitionID[%d] replicaID[%v] not active peer[%v]",
+							s.raftFsm.id, id, p.peer)
+						logger.Error(errMsg)
+						s.mStatus.replicasErrCnt[id] = 0
+					}
+				}
+			}
+		}
+	}
+}
+
 func (s *raft) tick() {
 	if s.restoringSnapshot.Get() {
 		return
@@ -446,6 +505,7 @@ func (s *raft) reciveMessage(m *proto.Message) {
 	case <-s.stopc:
 	case s.recvc <- m:
 	default:
+		logger.Warn(fmt.Sprintf("raft[%v] discard message(%v)", s.raftConfig.ID, m.ToString()))
 		return
 	}
 }

raft_fsm.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -21,6 +22,7 @@ import (
 
 	"github.com/tiglabs/raft/logger"
 	"github.com/tiglabs/raft/proto"
+	"time"
 )
 
 // NoLeader is a placeholder nodeID used when there is no leader.
@@ -130,9 +132,20 @@ func newRaftFsm(config *Config, raftConfig *RaftConfig) (*raftFsm, error) {
 		logger.Debug("newRaft[%v] [peers: [%s], term: %d, commit: %d, applied: %d, lastindex: %d, lastterm: %d]",
 			r.id, strings.Join(peerStrs, ","), r.term, r.raftLog.committed, r.raftLog.applied, r.raftLog.lastIndex(), r.raftLog.lastTerm())
 	}
+	go r.doRandomSeed()
 	return r, nil
 }
 
+func (r *raftFsm) doRandomSeed() {
+	ticker := time.Tick(time.Duration(rand.Intn(5)) * time.Second)
+	for {
+		select {
+		case <-ticker:
+			r.rand.Seed(time.Now().UnixNano())
+		}
+	}
+}
+
 // raft main method
 func (r *raftFsm) Step(m *proto.Message) {
 	if m.Type == proto.LocalMsgHup {
@@ -207,7 +220,7 @@ func (r *raftFsm) loadState(state proto.HardState) error {
 }
 
 func (r *raftFsm) recoverCommit() error {
-	for r.raftLog.applied < r.raftLog.committed {
+	for r.raftLog.applied <= r.raftLog.committed {
 		committedEntries := r.raftLog.nextEnts(64 * MB)
 		for _, entry := range committedEntries {
 			r.raftLog.appliedTo(entry.Index)
@@ -230,6 +243,9 @@ func (r *raftFsm) recoverCommit() error {
 				r.applyConfChange(cc)
 			}
 		}
+		if r.raftLog.applied == r.raftLog.committed {
+			break
+		}
 	}
 	return nil
 }
@@ -337,7 +353,10 @@ func (r *raftFsm) reset(term, lasti uint64, isLeader bool) {
 }
 
 func (r *raftFsm) resetRandomizedElectionTimeout() {
-	r.randElectionTick = r.config.ElectionTick + r.rand.Intn(r.config.ElectionTick)
+	randTick := r.rand.Intn(r.config.ElectionTick)
+	r.randElectionTick = r.config.ElectionTick + randTick
+	logger.Debug("raft[%v] random election timeout randElectionTick=%v, config.ElectionTick=%v, randTick=%v", r.id,
+		r.randElectionTick, r.config.ElectionTick, randTick)
 }
 
 func (r *raftFsm) pastElectionTimeout() bool {

raft_fsm_candidate.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

raft_fsm_follower.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

raft_fsm_leader.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -28,7 +29,7 @@ func (r *raftFsm) becomeLeader() {
 	if r.state == stateFollower {
 		panic(AppPanicError(fmt.Sprintf("[raft->becomeLeader][%v] invalid transition [follower -> leader].", r.id)))
 	}
-
+	r.recoverCommit()
 	lasti := r.raftLog.lastIndex()
 	r.step = stepLeader
 	r.reset(r.term, lasti, true)

raft_fsm_state.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.

raft_log.go

@@ -1,4 +1,5 @@
 // Copyright 2015 The etcd Authors
+// Modified work copyright 2018 The tiglabs Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.