From 44963061d36c13f97afa48250dd23d396313ad85 Mon Sep 17 00:00:00 2001 From: Wu Huocheng Date: Tue, 19 Aug 2025 16:54:22 +0800 Subject: [PATCH] fix(master): fix the planrun mistake.#1000286609 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 --- authnode/keystore_fsm.go | 6 +++--- datanode/data_partition_repair.go | 8 ++++---- datanode/partition.go | 4 ++-- datanode/wrap_operator.go | 6 +++--- master/admin_task_manager.go | 8 ++++---- master/cluster.go | 10 ++-------- master/cluster_balance.go | 3 ++- master/metadata_fsm.go | 8 ++++---- master/topology.go | 6 +++++- metanode/partition_delete_extents.go | 4 ++-- metanode/partition_free_list.go | 4 ++-- metanode/partition_fsm.go | 28 +++++++++++++++------------- util/exporter/alarm.go | 5 +++-- 13 files changed, 51 insertions(+), 49 deletions(-) diff --git a/authnode/keystore_fsm.go b/authnode/keystore_fsm.go index 489e5c4de..3879ab965 100644 --- a/authnode/keystore_fsm.go +++ b/authnode/keystore_fsm.go @@ -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 } diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 2397572af..a069590cd 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -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 } } diff --git a/datanode/partition.go b/datanode/partition.go index f520001ce..4b2802367 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -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 } diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index e84b0d0cb..aa5d669fe 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -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", diff --git a/master/admin_task_manager.go b/master/admin_task_manager.go index 4f121cbec..3e82aec94 100644 --- a/master/admin_task_manager.go +++ b/master/admin_task_manager.go @@ -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 diff --git a/master/cluster.go b/master/cluster.go index 759facb05..bd8608814 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -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 } } diff --git a/master/cluster_balance.go b/master/cluster_balance.go index ca9887448..00d5454ff 100644 --- a/master/cluster_balance.go +++ b/master/cluster_balance.go @@ -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 diff --git a/master/metadata_fsm.go b/master/metadata_fsm.go index 106e1ae56..69e6b6290 100644 --- a/master/metadata_fsm.go +++ b/master/metadata_fsm.go @@ -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 } diff --git a/master/topology.go b/master/topology.go index 9e89133bd..e2e211890 100644 --- a/master/topology.go +++ b/master/topology.go @@ -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 diff --git a/metanode/partition_delete_extents.go b/metanode/partition_delete_extents.go index 0a4f5b4d4..e718f745a 100644 --- a/metanode/partition_delete_extents.go +++ b/metanode/partition_delete_extents.go @@ -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) } }() diff --git a/metanode/partition_free_list.go b/metanode/partition_free_list.go index 06466871b..c729464cc 100644 --- a/metanode/partition_free_list.go +++ b/metanode/partition_free_list.go @@ -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) } }() diff --git a/metanode/partition_fsm.go b/metanode/partition_fsm.go index 4b0dac32e..ae66f9c0b 100644 --- a/metanode/partition_fsm.go +++ b/metanode/partition_fsm.go @@ -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 { diff --git a/util/exporter/alarm.go b/util/exporter/alarm.go index 7b7b67373..0e9c239ef 100644 --- a/util/exporter/alarm.go +++ b/util/exporter/alarm.go @@ -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) }