diff --git a/master/cluster.go b/master/cluster.go index 6891b8c24..83e2b4c24 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -1261,6 +1261,25 @@ 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 { + var dps *DataPartitionMap + 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() @@ -2378,6 +2397,94 @@ 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/topology.go b/master/topology.go index 0e4861d0c..736c75dad 100644 --- a/master/topology.go +++ b/master/topology.go @@ -30,7 +30,7 @@ import ( ) type rsManager struct { - // nodeType NodeType + nodeType NodeType nodes *sync.Map zoneIndexForNode int // zones []*Zone @@ -2184,6 +2184,7 @@ func (l *DecommissionDataPartitionList) traverse(c *Cluster) { log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for success", l.nsId, l, dp.PartitionID) } + auditlog.LogMasterOp("TraverseDataPartition", msg, err) } else if dp.IsDecommissionFailed() { remove := false if !dp.tryRollback(c) { diff --git a/master/vol_test.go b/master/vol_test.go index 33f8ced44..2956d236d 100644 --- a/master/vol_test.go +++ b/master/vol_test.go @@ -115,7 +115,7 @@ func TestCreateColdVol(t *testing.T) { require.EqualValues(t, defaultReplicaNum, vol.dpReplicaNum) require.EqualValues(t, 0, vol.domainId) - delVol(volName, t) + delVol(volName1, t) time.Sleep(30 * time.Second) req[nameKey] = volName2 diff --git a/metanode/api_handler.go b/metanode/api_handler.go index 3777338e2..a7aad164e 100644 --- a/metanode/api_handler.go +++ b/metanode/api_handler.go @@ -459,7 +459,7 @@ func (m *MetaNode) getExtentsByInodeHandler(w http.ResponseWriter, p.StartT = time.Now().UnixNano() p.ReqID = proto.GenerateRequestID() p.Opcode = proto.OpMetaExtentsList - p.PartitionID = pid + p.PartitionID = pid.V err = p.MarshalData(req) if err != nil { resp.Code = http.StatusInternalServerError @@ -742,11 +742,12 @@ func (m *MetaNode) getDirectoryHandler(w http.ResponseWriter, r *http.Request) { return } - pIno, err := strconv.ParseUint(r.FormValue("parentIno"), 10, 64) + p1, err := strconv.ParseUint(r.FormValue("parentIno"), 10, 64) if err != nil { resp.Msg = err.Error() return } + pid.V = p1 verSeq, err := m.getRealVerSeq(w, r) if err != nil { diff --git a/metanode/const.go b/metanode/const.go index 0ac3334d6..d48552db5 100644 --- a/metanode/const.go +++ b/metanode/const.go @@ -247,12 +247,8 @@ const ( // TODO: to remove unused by golangci var ( - _ = opFSMDelVer _ = opFSMDeletePartition - _ = opFSMSentToChanV1 - _ = opFSMStoreTickV1 _ = opFSMUpdateSummaryInfo - _ = (*metaPartition).doDeleteMarkedInodes _ = (*Dentry).getLastestVer _ = (*Inode).isEkInRefMap _ = (*metaPartition).notifyRaftFollowerToFreeInodes diff --git a/proto/model.go b/proto/model.go index 586a58afb..6db02c89c 100644 --- a/proto/model.go +++ b/proto/model.go @@ -396,15 +396,6 @@ type DataDecommissionProgress struct { ResidualDps []IgnoreDecommissionDP } -type DataDecommissionProgress struct { - Status uint32 - StatusMessage string - Progress string - FailedDps []FailedDpInfo - IgnoreDps []IgnoreDecommissionDP - ResidualDps []IgnoreDecommissionDP -} - type DiskInfo struct { NodeId uint64 Address string diff --git a/util/string.go b/util/string.go index b8fe95a01..733d59d80 100644 --- a/util/string.go +++ b/util/string.go @@ -230,45 +230,5 @@ func String2Any(str string, pvalue interface{}) error { if err != nil { return err } - - switch v := pvalue.(type) { - case *string: - *v = val.(string) - case *bool: - *v = val.(bool) - - case *int: - *v = int(val.(int64)) - case *int8: - *v = int8(val.(int64)) - case *int16: - *v = int16(val.(int64)) - case *int32: - *v = int32(val.(int64)) - case *int64: - *v = int64(val.(int64)) - - case *uint: - *v = uint(val.(uint64)) - case *uint8: - *v = uint8(val.(uint64)) - case *uint16: - *v = uint16(val.(uint64)) - case *uint32: - *v = uint32(val.(uint64)) - case *uint64: - *v = uint64(val.(uint64)) - - case *float32: - *v = float32(val.(float64)) - case *float64: - *v = float64(val.(float64)) - case *complex64: - *v = complex64(val.(complex128)) - case *complex128: - *v = complex128(val.(complex128)) - default: - return fmt.Errorf("unknown type %v of %s %v", v, str, pvalue) - } return nil }