Merge pull request #36 from absolute8511/raft-learner

Raft learner

node/namespace.go

@@ -113,7 +113,7 @@ func (nn *NamespaceNode) Start(forceStandaloneCluster bool) error {
 }
 
 func (nn *NamespaceNode) TransferMyLeader(to uint64, toRaftID uint64) error {
-	waitTimeout := time.Duration(nn.Node.machineConfig.ElectionTick)*time.Duration(nn.Node.machineConfig.TickMs)* time.Millisecond
+	waitTimeout := time.Duration(nn.Node.machineConfig.ElectionTick) * time.Duration(nn.Node.machineConfig.TickMs) * time.Millisecond
 	ctx, cancel := context.WithTimeout(context.Background(), waitTimeout)
 	defer cancel()
 	oldLeader := nn.Node.rn.Lead()
@@ -399,12 +399,12 @@ func (nsm *NamespaceMgr) GetNamespaceNodeFromGID(gid uint64) *NamespaceNode {
 	defer nsm.mutex.RUnlock()
 	gn, ok := nsm.groups[gid]
 	if !ok {
-		nodeLog.Errorf("group name not found %v ", gid)
+		nodeLog.Debugf("group name not found %v ", gid)
 		return nil
 	}
 	kv, ok := nsm.kvNodes[gn]
 	if !ok {
-		nodeLog.Errorf("kv namespace not found %v ", gn)
+		nodeLog.Infof("kv namespace not found %v ", gn)
 		return nil
 	}
 

node/raft.go

@@ -29,14 +29,14 @@ import (
 
 	"github.com/absolute8511/ZanRedisDB/common"
 	"github.com/absolute8511/ZanRedisDB/pkg/fileutil"
+	"github.com/absolute8511/ZanRedisDB/pkg/idutil"
+	"github.com/absolute8511/ZanRedisDB/pkg/types"
 	"github.com/absolute8511/ZanRedisDB/raft"
 	"github.com/absolute8511/ZanRedisDB/raft/raftpb"
 	"github.com/absolute8511/ZanRedisDB/snap"
 	"github.com/absolute8511/ZanRedisDB/transport/rafthttp"
 	"github.com/absolute8511/ZanRedisDB/wal"
 	"github.com/absolute8511/ZanRedisDB/wal/walpb"
-	"github.com/absolute8511/ZanRedisDB/pkg/idutil"
-	"github.com/absolute8511/ZanRedisDB/pkg/types"
 	"golang.org/x/net/context"
 )
 
@@ -611,7 +611,7 @@ func (rc *raftNode) beginSnapshot(snapi uint64, confState raftpb.ConfState) erro
 	if err != nil {
 		return err
 	}
-	rc.Infof("get snapshot object done: %v", snapi)
+	rc.Infof("get snapshot object done: %v, state: %v", snapi, confState.String())
 
 	rc.wgAsync.Add(1)
 	go func() {

raft/node.go

@@ -350,11 +350,11 @@ func (n *node) run(r *raft) {
 		case mdrop := <-n.recvc:
 			m := mdrop.m
 			// filter out response message from unknown From.
-			from, ok := r.prs[m.From]
-			if ok || !IsResponseMsg(m.Type) {
+			from := r.getProgress(m.From)
+			if from != nil || !IsResponseMsg(m.Type) {
 				if m.Type == pb.MsgTransferLeader {
 					if m.FromGroup.NodeId == 0 {
-						if !ok {
+						if from == nil {
 							if m.From == r.id {
 								m.FromGroup = r.group
 							} else {
@@ -367,8 +367,8 @@ func (n *node) run(r *raft) {
 						}
 					}
 					if m.ToGroup.NodeId == 0 {
-						g, ok := r.prs[m.To]
-						if !ok {
+						pr := r.getProgress(m.To)
+						if pr == nil {
 							if m.To == r.id {
 								m.ToGroup = r.group
 							} else {
@@ -377,13 +377,13 @@ func (n *node) run(r *raft) {
 								continue
 							}
 						} else {
-							m.ToGroup = g.group
+							m.ToGroup = pr.group
 						}
 					}
 				} else {
 					// if we missing the peer node group info, try update it from
 					// raft message
-					if ok && from.group.NodeId == 0 && m.FromGroup.NodeId > 0 &&
+					if from != nil && from.group.NodeId == 0 && m.FromGroup.NodeId > 0 &&
 						m.FromGroup.GroupId == r.group.GroupId {
 						from.group = m.FromGroup
 					}
@@ -402,6 +402,8 @@ func (n *node) run(r *raft) {
 			switch cc.Type {
 			case pb.ConfChangeAddNode:
 				r.addNode(cc.ReplicaID, cc.NodeGroup)
+			case pb.ConfChangeAddLearnerNode:
+				r.addLearner(cc.ReplicaID, cc.NodeGroup)
 			case pb.ConfChangeRemoveNode:
 				// block incoming proposal when local node is
 				// removed

raft/progress.go

@@ -16,6 +16,7 @@ package raft
 
 import (
 	"fmt"
+
 	pb "github.com/absolute8511/ZanRedisDB/raft/raftpb"
 )
 
@@ -82,6 +83,8 @@ type Progress struct {
 
 	// group indicate which node the raft follower is belonged
 	group pb.Group
+	// IsLearner is true if this progress is tracked for a learner.
+	IsLearner bool
 }
 
 func (pr *Progress) resetState(state ProgressStateType) {