diff --git a/.semgrep.yml b/.semgrep.yml index 7e8a42d13..b03de910a 100644 --- a/.semgrep.yml +++ b/.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 \ No newline at end of file + severity: WARNING diff --git a/blobstore/common/profile/doc.go b/blobstore/common/profile/doc.go index eafd5797a..e91db6be6 100644 --- a/blobstore/common/profile/doc.go +++ b/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) diff --git a/master/cluster.go b/master/cluster.go index 443bfd9e3..6f4f9bfce 100644 --- a/master/cluster.go +++ b/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; diff --git a/master/data_partition.go b/master/data_partition.go index ec5a7f6f3..5688add28 100644 --- a/master/data_partition.go +++ b/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] - partition.Replicas = append(partition.Replicas, replicasAfter...) + tmp := partition.Replicas[:index] + 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) } diff --git a/master/disk_manager.go b/master/disk_manager.go index 3b4568741..f6af2f4a7 100644 --- a/master/disk_manager.go +++ b/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 diff --git a/master/vol.go b/master/vol.go index cea679b61..4c4239f7f 100644 --- a/master/vol.go +++ b/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 diff --git a/proto/errors.go b/proto/errors.go index e30fa3a79..187360946 100644 --- a/proto/errors.go +++ b/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") diff --git a/sdk/data/stream/extent_handler.go b/sdk/data/stream/extent_handler.go index f38fdc1b4..fd2fa69d7 100644 --- a/sdk/data/stream/extent_handler.go +++ b/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