fix(master): assess replica[0] need dp lock

Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>

.semgrep.yml

@@ -8,6 +8,7 @@ rules:
         - '*.go'
     patterns:
       - pattern-regex: '(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)'
+      - pattern-not-regex: '127\.0\.0\.\d+'
       - pattern-not-regex: '10\.\d+\.\d+.\d+'
       - pattern-not-regex: '192\.168\.\d+.\d+'
       - pattern-not-regex: '192\.0\.2\.\d+'    # 192.0.2.0/24 (TEST-NET-1, rfc5737)
@@ -23,4 +24,4 @@ rules:
         - '*.go'
     patterns:
       - pattern-regex: '(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))'
-    severity: WARNING
+    severity: WARNING

blobstore/common/profile/doc.go

@@ -30,12 +30,12 @@
 //	}
 //
 //	func main() {
-//	 	ph := profile.NewProfileHandler("127.0.0.1:8888")
+//	 	ph := profile.NewProfileHandler("192.0.2.1:8888")
 //			httpServer := &http.Server{
-//				Addr:    "127.0.0.1:8888",
+//				Addr:    "192.0.2.1:8888",
 //				Handler: rpc.MiddlewareHandlerWith(rpc.DefaultRouter, ph),
 //			}
-//			log.Info("Server is running at", "127.0.0.1:8888")
+//			log.Info("Server is running at", "192.0.2.1:8888")
 //			go func() {
 //				err = httpServer.ListenAndServe()
 //				require.NoError(t, err)

master/cluster.go

@@ -857,32 +857,6 @@ func (c *Cluster) checkMetaPartitions() {
 	}
 }
 
-// move to partition.checkReplicaMeta
-func (c *Cluster) scheduleToReduceReplicaNum() {
-	go func() {
-		for {
-			if c.partition != nil && c.partition.IsRaftLeader() {
-				c.checkVolReduceReplicaNum()
-			}
-			time.Sleep(5 * time.Minute)
-		}
-	}()
-}
-
-func (c *Cluster) checkVolReduceReplicaNum() {
-	defer func() {
-		if r := recover(); r != nil {
-			log.LogWarnf("checkVolReduceReplicaNum occurred panic,err[%v]", r)
-			WarnBySpecialKey(fmt.Sprintf("%v_%v_scheduling_job_panic", c.Name, ModuleName),
-				"checkVolReduceReplicaNum occurred panic")
-		}
-	}()
-	vols := c.allVols()
-	for _, vol := range vols {
-		vol.checkReplicaNum(c)
-	}
-}
-
 func (c *Cluster) getInvalidIDNodes() (nodes []*InvalidNodeView) {
 	metaNodes := c.getNotConsistentIDMetaNodes()
 	nodes = append(nodes, metaNodes...)
@@ -891,36 +865,6 @@ func (c *Cluster) getInvalidIDNodes() (nodes []*InvalidNodeView) {
 	return
 }
 
-// move to partition.checkReplicaMeta
-func (c *Cluster) scheduleToCheckDataReplicas() {
-	go func() {
-		for {
-			if c.checkDataReplicasEnable {
-				if c.partition != nil && c.partition.IsRaftLeader() {
-					c.checkDataReplicas()
-				}
-			}
-			time.Sleep(1 * time.Minute)
-		}
-	}()
-}
-
-func (c *Cluster) checkDataReplicas() {
-	lackReplicaDataPartitions, _ := c.checkLackReplicaAndHostDataPartitions()
-	if len(lackReplicaDataPartitions) == 0 {
-		return
-	}
-
-	successCnt := 0
-	for _, dp := range lackReplicaDataPartitions {
-		if success, _ := c.autoAddDataReplica(dp); success {
-			successCnt += 1
-		}
-	}
-	failCnt := len(lackReplicaDataPartitions) - successCnt
-	log.LogInfof("action[checkDataReplicas] autoAddDataReplica successCnt[%v], failedCnt[%v]", successCnt, failCnt)
-}
-
 func (c *Cluster) getNotConsistentIDMetaNodes() (metaNodes []*InvalidNodeView) {
 	metaNodes = make([]*InvalidNodeView, 0)
 	c.metaNodes.Range(func(key, value interface{}) bool {
@@ -1189,24 +1133,6 @@ func (c *Cluster) checkInactiveDataNodes() (inactiveDataNodes []string, err erro
 	return
 }
 
-func (c *Cluster) checkLackReplicaAndHostDataPartitions() (lackReplicaDataPartitions []*DataPartition, err error) {
-	lackReplicaDataPartitions = make([]*DataPartition, 0)
-	vols := c.copyVols()
-	var ids []uint64
-	for _, vol := range vols {
-		dps := vol.dataPartitions
-		for _, dp := range dps.partitions {
-			if dp.ReplicaNum > uint8(len(dp.Hosts)) && len(dp.Hosts) == len(dp.Replicas) && (dp.IsDecommissionInitial() || dp.IsRollbackFailed()) {
-				lackReplicaDataPartitions = append(lackReplicaDataPartitions, dp)
-				ids = append(ids, dp.PartitionID)
-			}
-		}
-	}
-	log.LogInfof("clusterID[%v] checkLackReplicaAndHostDataPartitions count:[%v] ids[%v]", c.Name,
-		len(lackReplicaDataPartitions), ids)
-	return
-}
-
 func (c *Cluster) checkLackReplicaDataPartitions() (lackReplicaDataPartitions []*DataPartition, err error) {
 	lackReplicaDataPartitions = make([]*DataPartition, 0)
 	vols := c.copyVols()
@@ -2291,94 +2217,6 @@ ERR:
 	return err
 }
 
-func (c *Cluster) autoAddDataReplica(dp *DataPartition) (success bool, err error) {
-	var (
-		targetHosts []string
-		newAddr     string
-		vol         *Vol
-		zone        *Zone
-		ns          *nodeSet
-	)
-	success = false
-
-	dp.RLock()
-
-	// not support
-	if dp.isSpecialReplicaCnt() {
-		dp.RUnlock()
-		return
-	}
-
-	dp.RUnlock()
-
-	// not support
-	if !proto.IsNormalDp(dp.PartitionType) {
-		return
-	}
-
-	var ok bool
-	if vol, ok = c.vols[dp.VolName]; !ok {
-		log.LogWarnf("action[autoAddDataReplica] clusterID[%v] vol[%v] partitionID[%v] vol not exist, PersistenceHosts:[%v]",
-			c.Name, dp.VolName, dp.PartitionID, dp.Hosts)
-		return
-	}
-
-	// not support
-	if c.isFaultDomain(vol) {
-		return
-	}
-
-	if vol.crossZone {
-		zones := dp.getZones()
-		if targetHosts, _, err = c.getHostFromNormalZone(TypeDataPartition, zones, nil, dp.Hosts, 1, 1, ""); err != nil {
-			goto errHandler
-		}
-	} else {
-		if zone, err = c.t.getZone(vol.zoneName); err != nil {
-			log.LogWarnf("action[autoAddDataReplica] clusterID[%v] vol[%v] partitionID[%v] zone not exist, PersistenceHosts:[%v]",
-				c.Name, dp.VolName, dp.PartitionID, dp.Hosts)
-			return
-		}
-		nodeSets := dp.getNodeSets()
-		if len(nodeSets) != 1 {
-			log.LogWarnf("action[autoAddDataReplica] clusterID[%v] vol[%v] partitionID[%v] the number of nodeSets is not one, PersistenceHosts:[%v]",
-				c.Name, dp.VolName, dp.PartitionID, dp.Hosts)
-			return
-		}
-		if ns, err = zone.getNodeSet(nodeSets[0]); err != nil {
-			goto errHandler
-		}
-		if targetHosts, _, err = ns.getAvailDataNodeHosts(dp.Hosts, 1); err != nil {
-			goto errHandler
-		}
-	}
-
-	newAddr = targetHosts[0]
-	if err = c.addDataReplica(dp, newAddr, false); err != nil {
-		goto errHandler
-	}
-
-	dp.Status = proto.ReadOnly
-	dp.isRecover = true
-	c.putBadDataPartitionIDs(nil, newAddr, dp.PartitionID)
-
-	dp.RLock()
-	c.syncUpdateDataPartition(dp)
-	dp.RUnlock()
-
-	log.LogInfof("action[autoAddDataReplica] clusterID[%v] vol[%v] partitionID[%v] auto add data replica success, newReplicaHost[%v], PersistenceHosts:[%v]",
-		c.Name, dp.VolName, dp.PartitionID, newAddr, dp.Hosts)
-	success = true
-	return
-
-errHandler:
-	if err != nil {
-		err = fmt.Errorf("clusterID[%v] vol[%v] partitionID[%v], err[%v]", c.Name, dp.VolName, dp.PartitionID, err)
-		log.LogErrorf("action[autoAddDataReplica] err %v", err)
-	}
-	return
-}
-
 // Decommission a data partition.
 // 1. Check if we can decommission a data partition. In the following cases, we are not allowed to do so:
 // - (a) a replica is not in the latest host list;

master/data_partition.go

@@ -280,7 +280,9 @@ func (partition *DataPartition) createTaskToCreateDataPartition(addr string, dat
 ) {
 	leaderSize := 0
 	if createType == proto.DecommissionedCreateDataPartition {
+		partition.RLock()
 		leaderSize = int(partition.Replicas[0].Used)
+		partition.RUnlock()
 	}
 
 	task = proto.NewAdminTask(proto.OpCreateDataPartition, addr, newCreateDataPartitionRequest(
@@ -385,8 +387,9 @@ func (partition *DataPartition) deleteReplicaByIndex(index int) {
 	msg := fmt.Sprintf("deleteReplicaByIndex dp %v  index:%v  locations :%v ", partition.PartitionID, index, replicaAddrs)
 	log.LogInfo(msg)
 	replicasAfter := partition.Replicas[index+1:]
-	partition.Replicas = partition.Replicas[:index]
+	tmp := partition.Replicas[:index]
-	partition.Replicas = append(partition.Replicas, replicasAfter...)
+	tmp = append(tmp, replicasAfter...)
+	partition.Replicas = tmp
 }
 
 func (partition *DataPartition) createLoadTasks() (tasks []*proto.AdminTask) {
@@ -840,100 +843,6 @@ func (partition *DataPartition) activeUsedSimilar() bool {
 	return minus < util.GB
 }
 
-func (partition *DataPartition) getToBeDecommissionHost(replicaNum int) (host string) {
-	partition.RLock()
-	defer partition.RUnlock()
-	// if decommission happened, remove DecommissionDstAddr
-	if partition.IsRollbackFailed() {
-		if partition.isSpecialReplicaCnt() {
-			// if new replica is created, but decommission is failed for target data node is down. roll back is failed
-			// at this time. when data node is recover, new created replica can still do repair work
-			host = partition.DecommissionDstAddr
-			if host == "" {
-				log.LogInfof("action[getToBeDecommissionHost]  partition %v cannot find decommission dst",
-					partition.PartitionID)
-			}
-			hostLen := len(partition.Hosts)
-			if hostLen <= 1 || hostLen <= replicaNum {
-				return
-			}
-			host = partition.Hosts[hostLen-1]
-		} else {
-			// if rollback is failed
-			log.LogInfof("action[getToBeDecommissionHost]  partition %v need to decommission %v",
-				partition.PartitionID, partition.DecommissionDstAddr)
-			host = partition.DecommissionDstAddr
-		}
-		return
-	}
-	// remove the last host if decommission status is DecommissionInitial
-	if partition.IsDecommissionInitial() {
-		hostLen := len(partition.Hosts)
-		if hostLen <= 1 || hostLen <= replicaNum {
-			return
-		}
-		host = partition.Hosts[hostLen-1]
-	}
-	// if decommission is running, return ""
-	return
-}
-
-func (partition *DataPartition) removeOneReplicaByHost(c *Cluster, host string) (err error) {
-	if err = c.removeDataReplica(partition, host, false, false); err != nil {
-		return
-	}
-
-	partition.RLock()
-	defer partition.RUnlock()
-
-	if partition.IsRollbackFailed() {
-		partition.DecommissionDstAddr = ""
-		c.syncUpdateDataPartition(partition)
-		log.LogWarnf("action[removeOneReplicaByHost]  partition %v reset DecommissionDstAddr", partition.PartitionID)
-		return
-	}
-	// oldReplicaNum := partition.ReplicaNum
-	// partition.ReplicaNum = partition.ReplicaNum - 1
-	//
-	// if err = c.syncUpdateDataPartition(partition); err != nil {
-	//	partition.ReplicaNum = oldReplicaNum
-	// }
-
-	return
-}
-
-func (partition *DataPartition) getNodeSets() (nodeSets []uint64) {
-	partition.RLock()
-	defer partition.RUnlock()
-	nodeSetMap := map[uint64]struct{}{}
-	for _, replica := range partition.Replicas {
-		if replica.dataNode == nil {
-			continue
-		}
-		nodeSetMap[replica.dataNode.NodeSetID] = struct{}{}
-	}
-	for nodeSet := range nodeSetMap {
-		nodeSets = append(nodeSets, nodeSet)
-	}
-	return
-}
-
-func (partition *DataPartition) getZones() (zones []string) {
-	partition.RLock()
-	defer partition.RUnlock()
-	zoneMap := map[string]struct{}{}
-	for _, replica := range partition.Replicas {
-		if replica.dataNode == nil {
-			continue
-		}
-		zoneMap[replica.dataNode.ZoneName] = struct{}{}
-	}
-	for zone := range zoneMap {
-		zones = append(zones, zone)
-	}
-	return
-}
-
 func (partition *DataPartition) getLiveZones(offlineAddr string) (zones []string) {
 	partition.RLock()
 	defer partition.RUnlock()
@@ -1735,7 +1644,7 @@ func (partition *DataPartition) TryAcquireDecommissionToken(c *Cluster) bool {
 	begin := time.Now()
 	defer func() {
 		log.LogDebugf("action[TryAcquireDecommissionToken] dp %v get token to %v consume(%v) err(%v) result(%v)",
-			partition.decommissionInfo(), partition.DecommissionDstAddr, time.Now().Sub(begin).String(), err, result)
+			partition.decommissionInfo(), partition.DecommissionDstAddr, time.Since(begin).String(), err, result)
 	}()
 
 	// the first time for dst addr not specify
@@ -2192,10 +2101,6 @@ func (partition *DataPartition) isPerformingDecommission(c *Cluster) bool {
 	return ns.decommissionDataPartitionList.hasDecommissionToken(partition.PartitionID)
 }
 
-func (partition *DataPartition) isPerformingRestoreReplica() bool {
-	return atomic.LoadUint32(&partition.RestoreReplica) == RestoreReplicaMetaRunning
-}
-
 func (partition *DataPartition) setRestoreReplicaStatus(status uint32) {
 	atomic.StoreUint32(&partition.RestoreReplica, status)
 }

master/disk_manager.go

@@ -83,14 +83,14 @@ func (c *Cluster) checkDiskRecoveryProgress() {
 				log.LogWarnf("[checkDiskRecoveryProgress] dp(%v) is discard, decommission successfully", partition.PartitionID)
 				continue
 			}
-			//if len(partition.Replicas) == 0 ||
+			// if len(partition.Replicas) == 0 ||
 			//	(!partition.isSpecialReplicaCnt() && len(partition.Replicas) < int(partition.ReplicaNum)) ||
 			//	(partition.isSpecialReplicaCnt() && len(partition.Replicas) > int(partition.ReplicaNum)) {
 			//	newBadDpIds = append(newBadDpIds, partitionID)
 			//	log.LogInfof("action[checkDiskRecoveryProgress] dp %v newBadDpIds [%v] replics %v conf replics num %v",
 			//		partition.PartitionID, newBadDpIds, len(partition.Replicas), int(partition.ReplicaNum))
 			//	continue
-			//}
+			// }
 
 			newReplica, _ := partition.getReplica(partition.DecommissionDstAddr)
 			if newReplica == nil {
@@ -172,7 +172,6 @@ func (c *Cluster) checkDiskRecoveryProgress() {
 					err = c.syncUpdateDataPartition(partition)
 					if err != nil {
 						log.LogErrorf("[checkDiskRecoveryProgress] update dp(%v) fail, err(%v)", partitionID, err)
-						err = nil
 					}
 					partition.RUnlock()
 					continue

master/vol.go

@@ -715,38 +715,6 @@ func (vol *Vol) isOkUpdateRepCnt() (ok bool, rsp []uint64) {
 	return ok, rsp
 }
 
-func (vol *Vol) checkReplicaNum(c *Cluster) {
-	if !vol.NeedToLowerReplica {
-		return
-	}
-	var err error
-	if proto.IsCold(vol.VolType) {
-		return
-	}
-
-	dps := vol.cloneDataPartitionMap()
-	cnt := 0
-	for _, dp := range dps {
-		host := dp.getToBeDecommissionHost(int(vol.dpReplicaNum))
-		if host == "" {
-			continue
-		}
-		if err = dp.removeOneReplicaByHost(c, host); err != nil {
-			if dp.isSpecialReplicaCnt() && len(dp.Hosts) > 1 {
-				log.LogWarnf("action[checkReplicaNum] removeOneReplicaByHost host [%v],vol[%v],err[%v]", host, vol.Name, err)
-				continue
-			}
-			log.LogErrorf("action[checkReplicaNum] removeOneReplicaByHost host [%v],vol[%v],err[%v]", host, vol.Name, err)
-			continue
-		}
-		cnt++
-		if cnt > 100 {
-			return
-		}
-	}
-	vol.NeedToLowerReplica = false
-}
-
 func (vol *Vol) checkMetaPartitions(c *Cluster) {
 	var tasks []*proto.AdminTask
 	metaPartitionInodeIdStep := gConfig.MetaPartitionInodeIdStep

proto/errors.go

@@ -29,7 +29,6 @@ var (
 	ErrMetaPartitionNotExists = errors.New("meta partition not exists")
 	ErrDataPartitionNotExists = errors.New("data partition not exists")
 	ErrDataNodeNotExists      = errors.New("data node not exists")
-	ErrDiskNotExists          = errors.New("disk not exists")
 	ErrMetaNodeNotExists      = errors.New("meta node not exists")
 	ErrDuplicateVol           = errors.New("duplicate vol")
 	ErrActiveDataNodesTooLess = errors.New("no enough active data node")

sdk/data/stream/extent_handler.go

@@ -22,14 +22,13 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/cubefs/cubefs/util/stat"
-
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/sdk/data/wrapper"
 	"github.com/cubefs/cubefs/sdk/meta"
 	"github.com/cubefs/cubefs/util"
 	"github.com/cubefs/cubefs/util/errors"
 	"github.com/cubefs/cubefs/util/log"
+	"github.com/cubefs/cubefs/util/stat"
 )
 
 // State machines