Enhance: add check logic to client/partitions check process, enhance lock protection

Signed-off-by: leonrayang <chl696@sina.com>

client/common/taskpool_test.go

@@ -174,7 +174,7 @@ func merge(cs []<-chan int) <-chan int {
 	out := make(chan int)
 
 	wg.Add(len(cs))
-	for i, _ := range cs {
+	for i := range cs {
 		cc := cs[i]
 		go func() {
 			for n := range cc {

master/api_limiter.go

@@ -40,7 +40,7 @@ func newApiLimiter() *ApiLimiter {
 }
 
 func (l *ApiLimiter) clear() {
-	for k, _ := range l.limiterInfos {
+	for k := range l.limiterInfos {
 		delete(l.limiterInfos, k)
 	}
 }

master/api_service.go

@@ -3457,14 +3457,11 @@ func (m *Server) getDataPartitions(w http.ResponseWriter, r *http.Request) {
 	log.LogInfof("action[getDataPartitions] tmp is leader[%v]", m.cluster.partition.IsRaftLeader())
 	if !m.cluster.partition.IsRaftLeader() {
 		var ok bool
-		m.cluster.followerReadManager.rwMutex.RLock()
-		if body, ok = m.cluster.followerReadManager.volDataPartitionsView[name]; !ok {
-			m.cluster.followerReadManager.rwMutex.RUnlock()
+		if body, ok = m.cluster.followerReadManager.getVolViewAsFollower(name); !ok {
 			log.LogErrorf("action[getDataPartitions] volume [%v] not get partitions info", name)
 			sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("follower volume info not found")))
 			return
 		}
-		m.cluster.followerReadManager.rwMutex.RUnlock()
 		send(w, r, body)
 		return
 	}
@@ -3719,6 +3716,22 @@ func (m *Server) changeMasterLeader(w http.ResponseWriter, r *http.Request) {
 	_ = sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
 }
 
+func (m *Server) OpFollowerPartitionsRead(w http.ResponseWriter, r *http.Request) {
+	var (
+		err            error
+		enableFollower bool
+	)
+	log.LogDebugf("OpFollowerPartitionsRead.")
+	if enableFollower, err = extractStatus(r); err != nil {
+		log.LogErrorf("OpFollowerPartitionsRead.err %v", err)
+		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
+	}
+	m.cluster.followerReadManager.needCheck = enableFollower
+
+	rstMsg := fmt.Sprintf(" OpFollowerPartitionsRead. set needCheck %v command sucess. ", enableFollower)
+	_ = sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
+}
+
 func genRespMessage(data []byte, req *proto.APIAccessReq, ts int64, key []byte) (message string, err error) {
 	var (
 		jresp []byte

master/cluster.go

@@ -15,6 +15,7 @@
 package master
 
 import (
+	"encoding/json"
 	"fmt"
 	"strings"
 	"sync"
@@ -73,6 +74,7 @@ type followerReadManager struct {
 	volDataPartitionsView map[string][]byte
 	status                map[string]bool
 	lastUpdateTick        map[string]time.Time
+	needCheck             bool
 	rwMutex               sync.RWMutex
 }
 
@@ -109,16 +111,67 @@ func (mgr *followerReadManager) updateVolViewFromLeader(key string, value []byte
 	mgr.rwMutex.Lock()
 	defer mgr.rwMutex.Unlock()
 
+	if !mgr.checkViewContent(key, value, true) {
+		log.LogErrorf("updateVolViewFromLeader. checkViewContent failed")
+		return
+	}
 	log.LogInfof("action[updateVolViewFromLeader] volume %v be updated, value len %v", key, len(value))
 	mgr.volDataPartitionsView[key] = value
 	mgr.status[key] = true
 	mgr.lastUpdateTick[key] = time.Now()
 }
 
+func (mgr *followerReadManager) checkViewContent(volName string, data []byte, isUpdate bool) (ok bool) {
+	if !isUpdate && !mgr.needCheck {
+		return true
+	}
+
+	reply := &struct {
+		Code int32           `json:"code"`
+		Msg  string          `json:"msg"`
+		Data json.RawMessage `json:"data"`
+	}{}
+
+	view := &proto.DataPartitionsView{}
+
+	log.LogDebugf("volName %v do check content", volName)
+
+	if err := json.Unmarshal(data, reply); err != nil {
+		log.LogErrorf("checkViewContent. umarshal error volName %V", volName)
+		return false
+	}
+	if err := json.Unmarshal(reply.Data, view); err != nil {
+		log.LogErrorf("checkViewContent. umarshal reply.Data error volName %V", volName)
+		return false
+	}
+
+	if len(view.DataPartitions) == 0 {
+		log.LogErrorf("checkViewContent. get nil partitions volName %V", volName)
+		return false
+	}
+	for i := 0; i < len(view.DataPartitions); i++ {
+		dp := view.DataPartitions[i]
+		if len(dp.Hosts) == 0 {
+			log.LogErrorf("checkViewContent. dp id %v, leader %v, status %v", dp.PartitionID, dp.LeaderAddr, dp.Status)
+			ok = false
+		}
+	}
+	log.LogDebugf("checkViewContent. volName %v dp cnt %v check pass", volName, len(view.DataPartitions))
+	return true
+}
+
 func (mgr *followerReadManager) getVolViewAsFollower(key string) (value []byte, ok bool) {
-	mgr.rwMutex.Lock()
-	defer mgr.rwMutex.Unlock()
-	value, ok = mgr.volDataPartitionsView[key]
+	mgr.rwMutex.RLock()
+	defer mgr.rwMutex.RUnlock()
+	if !mgr.status[key] {
+		return
+	}
+	value, _ = mgr.volDataPartitionsView[key]
+	ok = mgr.checkViewContent(key, value, false)
+	if !ok {
+		log.LogWarnf("getVolViewAsFollower. set volume %v not worked!", key)
+		mgr.status[key] = false
+	}
 	return
 }
 
@@ -2082,6 +2135,9 @@ func (c *Cluster) removeHostMember(dp *DataPartition, removePeer proto.Peer) (er
 		}
 		newPeers = append(newPeers, peer)
 	}
+
+	dp.Lock()
+	defer dp.Unlock()
 	if err = dp.update("removeDataPartitionRaftMember", dp.VolName, newPeers, newHosts, c); err != nil {
 		return
 	}

master/data_partition.go

@@ -645,6 +645,10 @@ func (partition *DataPartition) getReplicaIndex(addr string) (index int, err err
 }
 
 func (partition *DataPartition) update(action, volName string, newPeers []proto.Peer, newHosts []string, c *Cluster) (err error) {
+	if len(newHosts) == 0 {
+		log.LogErrorf("update. action[%v] update partition[%v] vol[%v] old host[%v]", action, partition.PartitionID, volName, partition.Hosts)
+		return
+	}
 	orgHosts := make([]string, len(partition.Hosts))
 	copy(orgHosts, partition.Hosts)
 	oldPeers := make([]proto.Peer, len(partition.Peers))
@@ -659,7 +663,7 @@ func (partition *DataPartition) update(action, volName string, newPeers []proto.
 	msg := fmt.Sprintf("action[%v] success,vol[%v] partitionID:%v "+
 		"oldHosts:%v newHosts:%v,oldPees[%v],newPeers[%v]",
 		action, volName, partition.PartitionID, orgHosts, partition.Hosts, oldPeers, partition.Peers)
-	log.LogInfo(msg)
+	log.LogWarnf(msg)
 	return
 }
 

master/data_partition_map.go

@@ -169,6 +169,10 @@ func (dpMap *DataPartitionMap) getDataPartitionsView(minPartitionID uint64) (dpR
 	dpMap.RLock()
 	defer dpMap.RUnlock()
 	for _, dp := range dpMap.partitionMap {
+		if len(dp.Hosts) == 0 {
+			log.LogErrorf("getDataPartitionsView. dp %v host nil", dp.PartitionID)
+			continue
+		}
 		if dp.PartitionID <= minPartitionID {
 			continue
 		}

master/http_server.go

@@ -64,7 +64,7 @@ func (m *Server) isFollowerRead(r *http.Request) (followerRead bool) {
 				return
 			}
 		}
-	} else if r.URL.Path == proto.AdminChangeMasterLeader {
+	} else if r.URL.Path == proto.AdminChangeMasterLeader || r.URL.Path == proto.AdminOpFollowerPartitionsRead {
 		followerRead = true
 	}
 	return
@@ -203,7 +203,9 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
 	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
 		Path(proto.AdminChangeMasterLeader).
 		HandlerFunc(m.changeMasterLeader)
-
+	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
+		Path(proto.AdminOpFollowerPartitionsRead).
+		HandlerFunc(m.OpFollowerPartitionsRead)
 	// node task response APIs
 	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
 		Path(proto.GetDataNodeTaskResponse).

proto/admin_proto.go

@@ -58,6 +58,7 @@ const (
 	AdminSetDpRdOnly                = "/admin/setDpRdOnly"
 	AdminDataPartitionChangeLeader  = "/dataPartition/changeleader"
 	AdminChangeMasterLeader         = "/master/changeleader"
+	AdminOpFollowerPartitionsRead   = "/master/opFollowerPartitionRead"
 	//graphql master api
 	AdminClusterAPI = "/api/cluster"
 	AdminUserAPI    = "/api/user"