gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(master): fix the planrun mistake.#1000286609

Browse Source 
Update the codes with reviewer's comments.
Update codes that is such as: log.LogErrorf(fmt.Sprintf.
remove the fmt.Sprint

Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>
This commit is contained in:
Wu Huocheng 2025-08-19 16:54:22 +08:00 committed by wu huocheng
parent 7858ae3d58
commit 44963061d3
13 changed files with 51 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
authnode/keystore_fsm.go
View File

@ -200,7 +200,7 @@ func (mf *KeystoreFsm) Snapshot() (proto.Snapshot, error) {
// ApplySnapshot implements the interface of raft.StateMachine
func (mf *KeystoreFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIterator) (err error) {
log.LogInfof(fmt.Sprintf("action[ApplySnapshot] begin,applied[%v]", mf.applied))
log.LogInfof("action[ApplySnapshot] begin,applied[%v]", mf.applied)
var data []byte
for err == nil {
if data, err = iterator.Next(); err != nil {
@ -218,10 +218,10 @@ func (mf *KeystoreFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIter
goto errHandler
}
mf.snapshotHandler()
log.LogInfof(fmt.Sprintf("action[ApplySnapshot] success,applied[%v]", mf.applied))
log.LogInfof("action[ApplySnapshot] success,applied[%v]", mf.applied)
return nil
errHandler:
log.LogError(fmt.Sprintf("action[ApplySnapshot] failed,err:%v", err.Error()))
log.LogErrorf("action[ApplySnapshot] failed,err:%v", err.Error())
return err
}

8
datanode/data_partition_repair.go
View File

@ -890,9 +890,9 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo,
return
}
if loopTimes%100 == 0 {
log.LogInfof(fmt.Sprintf("action[streamRepairExtent] dp %v extent %v fix(%v) start fix from (%v)"+
log.LogInfof("action[streamRepairExtent] dp %v extent %v fix(%v) start fix from (%v)"+
" remoteSize(%v)localSize(%v) reply(%v).", dp.partitionID, remoteExtentInfo.FileID, localExtentInfo.FileID, remoteExtentInfo.String(),
dstOffset, currFixOffset, reply.GetUniqueLogId()))
dstOffset, currFixOffset, reply.GetUniqueLogId())
}
loopTimes++
actualCrc := crc32.ChecksumIEEE(reply.GetData()[:reply.GetSize()])
@ -967,9 +967,9 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo,
hasRecoverySize += uint64(reply.GetSize())
currFixOffset += uint64(reply.GetSize())
if currFixOffset >= dstOffset {
log.LogWarnf(fmt.Sprintf("action[streamRepairExtent] dp %v extent(%v) start fix from (%v)"+
log.LogWarnf("action[streamRepairExtent] dp %v extent(%v) start fix from (%v)"+
" remoteSize(%v)localSize(%v) reply(%v) size(%v) cost(%v)millseconds.", dp.partitionID, localExtentInfo.FileID, remoteExtentInfo.String(),
remoteExtentInfo.Size, currFixOffset, reply.GetUniqueLogId(), currFixOffset-starFixOffset, time.Since(begin).Milliseconds()))
remoteExtentInfo.Size, currFixOffset, reply.GetUniqueLogId(), currFixOffset-starFixOffset, time.Since(begin).Milliseconds())
break
}
}

4
datanode/partition.go
View File

@ -1088,8 +1088,8 @@ func (dp *DataPartition) updateReplicas(isForce bool) (err error) {
dp.isLeader = isLeader
// dp.replicas = replicas
dp.intervalToUpdateReplicas = time.Now().Unix()
log.LogInfof(fmt.Sprintf("ActionUpdateReplicationHosts partiton(%v), force(%v) isLeader(%v)",
dp.partitionID, isForce, isLeader))
log.LogInfof("ActionUpdateReplicationHosts partiton(%v), force(%v) isLeader(%v)",
dp.partitionID, isForce, isLeader)
return
}

6
datanode/wrap_operator.go
View File

@ -690,7 +690,7 @@ func (s *DataNode) handlePacketToDeleteDataPartition(p *repl.Packet) {
return
}
request := &proto.DeleteDataPartitionRequest{}
log.LogInfof(fmt.Sprintf("action[handlePacketToDeleteDataPartition] request %v ", request))
log.LogInfof("action[handlePacketToDeleteDataPartition] request %v ", request)
if task.OpCode == proto.OpDeleteDataPartition {
bytes, _ := json.Marshal(task.Request)
p.AddMesgLog(string(bytes))
@ -707,7 +707,7 @@ func (s *DataNode) handlePacketToDeleteDataPartition(p *repl.Packet) {
err = errors.Trace(err, "delete DataPartition failed,PartitionID(%v)", request.PartitionId)
log.LogErrorf("action[handlePacketToDeleteDataPartition] err(%v).", err)
} else {
log.LogInfof(fmt.Sprintf("action[handlePacketToDeleteDataPartition] %v success", request.PartitionId))
log.LogInfof("action[handlePacketToDeleteDataPartition] %v success", request.PartitionId)
}
auditlog.LogDataNodeOp("DeleteDataPartition", fmt.Sprintf("%v is deleted", request.PartitionId), err)
}
@ -849,7 +849,7 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
return
}
log.LogInfof(fmt.Sprintf("[handleBatchMarkDeletePacket] recive DeleteExtent (%v) from (%v)", ext, c.RemoteAddr().String()))
log.LogInfof("[handleBatchMarkDeletePacket] recive DeleteExtent (%v) from (%v)", ext, c.RemoteAddr().String())
partition.disk.diskLimit(OpDelete, 0, func() {
if storage.IsTinyExtent(ext.ExtentId) || ext.IsSnapshotDeletion {
log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v), tinyExtent or snapDeletion",

8
master/admin_task_manager.go
View File

@ -94,8 +94,8 @@ func (sender *AdminTaskManager) getToBeDeletedTasks() (delTasks []*proto.AdminTa
for _, task := range sender.TaskMap {
if task.CheckTaskTimeOut() {
log.LogWarnf(fmt.Sprintf("clusterID[%v] %v has no response until time out",
sender.clusterID, task.ID))
log.LogWarnf("clusterID[%v] %v has no response until time out",
sender.clusterID, task.ID)
if task.SendTime > 0 {
Warn(sender.clusterID, fmt.Sprintf("clusterID[%v] %v has no response until time out",
sender.clusterID, task.ID))
@ -194,7 +194,7 @@ func (sender *AdminTaskManager) sendAdminTask(task *proto.AdminTask, conn net.Co
if err = packet.ReadFromConnWithVer(conn, proto.ReadDeadlineTime); err != nil {
return errors.Trace(err, "action[sendAdminTask],ReadFromConn failed task:%v", task.ID)
}
log.LogDebugf(fmt.Sprintf("action[sendAdminTask] sender task:%v success", task.ToString()))
log.LogDebugf("action[sendAdminTask] sender task:%v success", task.ToString())
sender.updateTaskInfo(task, true)
return nil
@ -225,7 +225,7 @@ func (sender *AdminTaskManager) syncSendAdminTask(task *proto.AdminTask) (packet
}
if packet.ResultCode != proto.OpOk {
err = fmt.Errorf("result code[%v],msg[%v]", packet.ResultCode, string(packet.Data))
log.LogInfof("action[syncSendAdminTask],task:%v,reqID[%v],err[%v],", task.ID, packet.ReqID, err)
log.LogErrorf("action[syncSendAdminTask],task:%v,reqID[%v],err[%v],", task.ID, packet.ReqID, err)
return
}
return packet, nil

10
master/cluster.go
View File

@ -2266,14 +2266,8 @@ func (c *Cluster) getHostFromNormalZone(nodeType uint32, excludeZones []string,
}
} else {
rsMgr = &c.t.metaTopology
if nodeType == TypeMetaPartition {
if zonesQualified, err = c.t.allocZonesForMetaNode(zoneNumNeed, replicaNum, excludeZones, proto.StoreModeMem); err != nil {
return
}
} else {
if zonesQualified, err = c.t.allocZonesForMetaNode(zoneNumNeed, replicaNum, excludeZones, proto.StoreModeRocksDb); err != nil {
return
}
if zonesQualified, err = c.t.allocZonesForMetaNode(zoneNumNeed, replicaNum, excludeZones, nodeType); err != nil {
return
}
}

3
master/cluster_balance.go
View File

@ -1124,6 +1124,7 @@ func (c *Cluster) RunMetaPartitionBalanceTask() error {
return err
}
c.PlanRun = true
go c.DoMetaPartitionBalanceTask(plan)
return nil
@ -1135,7 +1136,6 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) {
err error
)
c.PlanRun = true
defer func() {
// clear the run flag.
c.PlanRun = false
@ -1526,6 +1526,7 @@ func (c *Cluster) RestartMetaPartitionBalanceTask() error {
return nil
}
c.PlanRun = true
go c.DoMetaPartitionBalanceTask(plan)
return nil

8
master/metadata_fsm.go
View File

@ -251,7 +251,7 @@ func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIter
}()
}
log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] begin,applied[%v]", mf.applied))
log.LogWarnf("action[ApplySnapshot] begin,applied[%v]", mf.applied)
var data []byte
var appliedIndex []byte
for err == nil {
@ -279,7 +279,7 @@ func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIter
}
if err = mf.store.Flush(); err != nil {
log.LogError(fmt.Sprintf("action[ApplySnapshot] Flush failed,err:%v", err.Error()))
log.LogErrorf("action[ApplySnapshot] Flush failed,err:%v", err.Error())
goto errHandler
}
@ -294,10 +294,10 @@ func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIter
}
mf.snapshotHandler()
log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] success,applied[%v]", mf.applied))
log.LogWarnf("action[ApplySnapshot] success,applied[%v]", mf.applied)
return nil
errHandler:
log.LogError(fmt.Sprintf("action[ApplySnapshot] failed,err:%v", err.Error()))
log.LogErrorf("action[ApplySnapshot] failed,err:%v", err.Error())
return err
}

6
master/topology.go
View File

@ -2713,7 +2713,7 @@ func (zone *Zone) canWriteForMetaNode(replicaNum uint8, storeMode proto.StoreMod
return
}
func (t *topology) allocZonesForMetaNode(zoneNum, replicaNum int, excludeZone []string, storeMode proto.StoreMode) (zones []*Zone, err error) {
func (t *topology) allocZonesForMetaNode(zoneNum, replicaNum int, excludeZone []string, nodeType uint32) (zones []*Zone, err error) {
if len(t.domainExcludeZones) > 0 {
zones = t.getDomainExcludeZones()
log.LogInfof("action[allocZonesForMetaNode] getDomainExcludeZones zones [%v]", t.domainExcludeZones)
@ -2729,6 +2729,10 @@ func (t *topology) allocZonesForMetaNode(zoneNum, replicaNum int, excludeZone []
}
candidateZones := make([]*Zone, 0)
demandWriteNodes := calculateDemandWriteNodes(zoneNum, replicaNum, false)
storeMode := proto.StoreModeMem
if nodeType == TypeMetaPartition {
storeMode = proto.StoreModeRocksDb
}
for i := 0; i < len(zones); i++ {
if t.metaTopology.zoneIndexForNode >= len(zones) {
t.metaTopology.zoneIndexForNode = 0

4
metanode/partition_delete_extents.go
View File

@ -84,7 +84,7 @@ func (mp *metaPartition) createExtentDeleteFile(prefix string, idx int64, fileLi
func (mp *metaPartition) appendDelExtentsToFile(fileList *synclist.SyncList) {
defer func() {
if r := recover(); r != nil {
log.LogErrorf(fmt.Sprintf("[metaPartition] appendDelExtentsToFile pid(%v) panic (%v)", mp.config.PartitionId, r))
log.LogErrorf("[metaPartition] appendDelExtentsToFile pid(%v) panic (%v)", mp.config.PartitionId, r)
}
}()
var (
@ -230,7 +230,7 @@ func (mp *metaPartition) batchDeleteExtentsByDp(dpId uint64, extents []*proto.De
func (mp *metaPartition) deleteExtentsFromList(fileList *synclist.SyncList) {
defer func() {
if r := recover(); r != nil {
log.LogErrorf(fmt.Sprintf("deleteExtentsFromList(%v) deleteExtentsFromList panic (%v)", mp.config.PartitionId, r))
log.LogErrorf("deleteExtentsFromList(%v) deleteExtentsFromList panic (%v)", mp.config.PartitionId, r)
}
}()

4
metanode/partition_free_list.go
View File

@ -349,8 +349,8 @@ func (mp *metaPartition) deleteMarkedInodes(inoSlice []uint64) {
defer func() {
if r := recover(); r != nil {
stack := string(debug.Stack())
log.LogErrorf(fmt.Sprintf("metaPartition(%v) deleteMarkedInodes panic (%v)\nstack:%v",
mp.config.PartitionId, r, stack))
log.LogErrorf("metaPartition(%v) deleteMarkedInodes panic (%v)\nstack:%v",
mp.config.PartitionId, r, stack)
}
}()

28
metanode/partition_fsm.go
View File

@ -1066,8 +1066,9 @@ func (mp *metaPartition) HandleFatalEvent(err *raft.FatalError) {
// HandleLeaderChange handles the leader changes.
func (mp *metaPartition) HandleLeaderChange(leader uint64) {
exporter.Warning(fmt.Sprintf("metaPartition(%v) changeLeader to (%v)", mp.config.PartitionId, leader))
log.LogDebugf(fmt.Sprintf("metaPartition(%v) changeLeader to (%v)", mp.config.PartitionId, leader))
msg := fmt.Sprintf("metaPartition(%v) changeLeader to (%v)", mp.config.PartitionId, leader)
exporter.Warning(msg)
log.LogDebugf(msg)
if mp.config.NodeId == leader {
localIp := mp.manager.metaNode.localAddr
if localIp == "" {
@ -1076,23 +1077,24 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) {
conn, err := net.DialTimeout("tcp", net.JoinHostPort(localIp, serverPort), time.Second)
if err != nil {
log.LogErrorf(fmt.Sprintf("HandleLeaderChange serverPort not exsit ,error %v", err))
exporter.Warning(fmt.Sprintf("mp[%v] HandleLeaderChange serverPort not exsit ,error %v", mp.config.PartitionId, err))
msg = fmt.Sprintf("mp[%v] HandleLeaderChange serverPort not exsit ,error %v", mp.config.PartitionId, err)
log.LogErrorf(msg)
exporter.Warning(msg)
go mp.raftPartition.TryToLeader(mp.config.PartitionId)
return
}
log.LogDebugf("[metaPartition] pid: %v HandleLeaderChange close conn %v, nodeId: %v, leader: %v",
msg = fmt.Sprintf("[metaPartition]mp[%v] HandleLeaderChange close conn %v, nodeId: %v, leader: %v",
mp.config.PartitionId, serverPort, mp.config.NodeId, leader)
exporter.Warning(fmt.Sprintf("[metaPartition]mp[%v] HandleLeaderChange close conn %v, nodeId: %v, leader: %v",
mp.config.PartitionId, serverPort, mp.config.NodeId, leader))
log.LogDebugf(msg)
exporter.Warning(msg)
conn.(*net.TCPConn).SetLinger(0)
conn.Close()
}
if mp.config.NodeId != leader {
log.LogDebugf("[metaPartition] pid: %v HandleLeaderChange become unleader nodeId: %v, leader: %v",
msg = fmt.Sprintf("[metaPartition] pid: %v HandleLeaderChange become unleader nodeId: %v, leader: %v",
mp.config.PartitionId, mp.config.NodeId, leader)
exporter.Warning(fmt.Sprintf("[metaPartition] pid: %v HandleLeaderChange become unleader nodeId: %v, leader: %v",
mp.config.PartitionId, mp.config.NodeId, leader))
log.LogDebugf(msg)
exporter.Warning(msg)
mp.storeChan <- &storeMsg{
command: stopStoreTick,
}
@ -1102,10 +1104,10 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) {
command: startStoreTick,
}
log.LogDebugf("[metaPartition] pid: %v HandleLeaderChange become leader conn %v, nodeId: %v, leader: %v",
msg = fmt.Sprintf("[metaPartition] pid: %v HandleLeaderChange become leader conn %v, nodeId: %v, leader: %v",
mp.config.PartitionId, serverPort, mp.config.NodeId, leader)
exporter.Warning(fmt.Sprintf("[metaPartition] pid: %v HandleLeaderChange become leader conn %v, nodeId: %v, leader: %v",
mp.config.PartitionId, serverPort, mp.config.NodeId, leader))
log.LogDebugf(msg)
exporter.Warning(msg)
if mp.config.Start == 0 && mp.config.Cursor == 0 {
id, err := mp.nextInodeID()
if err != nil {

5
util/exporter/alarm.go
View File

@ -62,6 +62,7 @@ func WarningRocksdbError(detail string) {
}
func WarningAppendKey(key, detail string) {
ump.Alarm(fmt.Sprintf("%v_%v_%v", clustername, modulename, key), detail)
log.LogCritical(fmt.Sprintf("%v_%v_%v", clustername, modulename, key), detail)
newKey := fmt.Sprintf("%v_%v_%v", clustername, modulename, key)
ump.Alarm(newKey, detail)
log.LogCritical(newKey, detail)
}