diff --git a/cli/cmd/fmt.go b/cli/cmd/fmt.go index 9b21d8149..3765123d2 100644 --- a/cli/cmd/fmt.go +++ b/cli/cmd/fmt.go @@ -638,6 +638,7 @@ func formatMetaPartitionInfo(partition *proto.MetaPartitionInfo) string { } sb.WriteString("\n") sb.WriteString("Peers :\n") + sb.WriteString(fmt.Sprintf("%v\n", formatPeerTableHeader())) for _, peer := range partition.Peers { sb.WriteString(fmt.Sprintf("%v\n", formatPeer(peer))) } @@ -916,14 +917,14 @@ func formatMetaReplica(indentation string, replica *proto.MetaReplicaInfo, rowTa return sb.String() } -var peerTableRowPattern = "%-6v %-18v" +var peerTableRowPattern = "%-6v %-18v %-12v %-12v" func formatPeerTableHeader() string { - return fmt.Sprintf(peerTableRowPattern, "ID", "PEER") + return fmt.Sprintf(peerTableRowPattern, "ID", "ADDR", "HEARTBEATPORT", "REPLICAPORT") } func formatPeer(peer proto.Peer) string { - return fmt.Sprintf(peerTableRowPattern, peer.ID, peer.Addr) + return fmt.Sprintf(peerTableRowPattern, peer.ID, peer.Addr, peer.HeartbeatPort, peer.ReplicaPort) } var dataNodeDetailTableRowPattern = "%-6v %-6v %-65v %-6v %-6v %-6v %-10v" diff --git a/datanode/partition.go b/datanode/partition.go index c778ab6c2..dd7ddc443 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -25,6 +25,7 @@ import ( "os" "path" "sort" + "strconv" "strings" "sync" "sync/atomic" @@ -416,6 +417,18 @@ func newDataPartition(dpCfg *dataPartitionCfg, disk *Disk, isCreate bool) (dp *D responseStatus: responseInitial, PersistApplyIdChan: make(chan PersistApplyIdRequest), } + + // during upgrade process, create partition request may lack raft ports info + defaultHeartbeatPort, defaultReplicaPort, err := partition.raftPort() + if err == nil { + for i := range partition.config.Peers { + if len(partition.config.Peers[i].ReplicaPort) == 0 || len(partition.config.Peers[i].HeartbeatPort) == 0 { + partition.config.Peers[i].ReplicaPort = strconv.FormatInt(int64(defaultReplicaPort), 10) + partition.config.Peers[i].HeartbeatPort = strconv.FormatInt(int64(defaultHeartbeatPort), 10) + } + } + } + atomic.StoreUint64(&partition.recoverErrCnt, 0) log.LogInfof("action[newDataPartition] dp %v replica num %v", partitionID, dpCfg.ReplicaNum) diff --git a/datanode/partition_raft.go b/datanode/partition_raft.go index ac93a7b13..7b38f525f 100644 --- a/datanode/partition_raft.go +++ b/datanode/partition_raft.go @@ -56,28 +56,28 @@ type dataPartitionCfg struct { ForbidWriteOpOfProtoVer0 bool } -//func (dp *DataPartition) raftPort() (heartbeat, replica int, err error) { -// raftConfig := dp.config.RaftStore.RaftConfig() -// heartbeatAddrSplits := strings.Split(raftConfig.HeartbeatAddr, ":") -// replicaAddrSplits := strings.Split(raftConfig.ReplicateAddr, ":") -// if len(heartbeatAddrSplits) != 2 { -// err = errors.New("illegal heartbeat address") -// return -// } -// if len(replicaAddrSplits) != 2 { -// err = errors.New("illegal replica address") -// return -// } -// heartbeat, err = strconv.Atoi(heartbeatAddrSplits[1]) -// if err != nil { -// return -// } -// replica, err = strconv.Atoi(replicaAddrSplits[1]) -// if err != nil { -// return -// } -// return -//} +func (dp *DataPartition) raftPort() (heartbeat, replica int, err error) { + raftConfig := dp.config.RaftStore.RaftConfig() + heartbeatAddrSplits := strings.Split(raftConfig.HeartbeatAddr, ":") + replicaAddrSplits := strings.Split(raftConfig.ReplicateAddr, ":") + if len(heartbeatAddrSplits) != 2 { + err = errors.New("illegal heartbeat address") + return + } + if len(replicaAddrSplits) != 2 { + err = errors.New("illegal replica address") + return + } + heartbeat, err = strconv.Atoi(heartbeatAddrSplits[1]) + if err != nil { + return + } + replica, err = strconv.Atoi(replicaAddrSplits[1]) + if err != nil { + return + } + return +} // StartRaft start raft instance when data partition start or restore. func (dp *DataPartition) StartRaft(isLoad bool) (err error) { @@ -92,9 +92,9 @@ func (dp *DataPartition) StartRaft(isLoad bool) (err error) { } var ( - heartbeatPort int - replicaPort int - peers []raftstore.PeerAddress + defaultHeartbeatPort int + defaultReplicaPort int + peers []raftstore.PeerAddress ) defer func() { if r := recover(); r != nil { @@ -109,13 +109,21 @@ func (dp *DataPartition) StartRaft(isLoad bool) (err error) { } }() - //if heartbeatPort, replicaPort, err = dp.raftPort(); err != nil { - // return - //} + if defaultHeartbeatPort, defaultReplicaPort, err = dp.raftPort(); err != nil { + return + } for _, peer := range dp.config.Peers { addr := strings.Split(peer.Addr, ":")[0] - heartbeatPort, _ = strconv.Atoi(peer.HeartbeatPort) - replicaPort, _ = strconv.Atoi(peer.ReplicaPort) + + heartbeatPort, perr := strconv.Atoi(peer.HeartbeatPort) + if perr != nil { + heartbeatPort = defaultHeartbeatPort + } + replicaPort, perr := strconv.Atoi(peer.ReplicaPort) + if perr != nil { + replicaPort = defaultReplicaPort + } + rp := raftstore.PeerAddress{ Peer: raftproto.Peer{ ID: peer.ID, @@ -409,16 +417,21 @@ func (dp *DataPartition) addRaftNode(req *proto.AddDataPartitionRaftMemberReques } var ( - heartbeatPort int - replicaPort int + defaultHeartbeatPort int + defaultReplicaPort int ) - heartbeatPort, err = strconv.Atoi(req.AddPeer.HeartbeatPort) - if err != nil { + if defaultHeartbeatPort, defaultReplicaPort, err = dp.raftPort(); err != nil { return } - replicaPort, err = strconv.Atoi(req.AddPeer.ReplicaPort) - if err != nil { + heartbeatPort, perr := strconv.Atoi(req.AddPeer.HeartbeatPort) + if perr != nil { + heartbeatPort = defaultHeartbeatPort + } + + replicaPort, perr := strconv.Atoi(req.AddPeer.ReplicaPort) + if perr != nil { + replicaPort = defaultReplicaPort return } diff --git a/master/api_args_parse.go b/master/api_args_parse.go index 7a7d5c88e..533acd98f 100644 --- a/master/api_args_parse.go +++ b/master/api_args_parse.go @@ -26,8 +26,6 @@ import ( "strings" "time" - "github.com/cubefs/cubefs/raftstore" - "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/compressor" @@ -177,14 +175,9 @@ func parseRequestForAddNode(r *http.Request) (nodeAddr, raftHeartbeatPort, raftR if zoneName = r.FormValue(zoneNameKey); zoneName == "" { zoneName = DefaultZoneName } - - if raftHeartbeatPort = r.FormValue(heartbeatPortKey); raftHeartbeatPort == "" { - raftHeartbeatPort = strconv.Itoa(raftstore.DefaultHeartbeatPort) - } - - if raftReplicaPort = r.FormValue(replicaPortKey); raftReplicaPort == "" { - raftReplicaPort = strconv.Itoa(raftstore.DefaultReplicaPort) - } + // for old version node registration, heartbeat port and replica port may be empty + raftHeartbeatPort = r.FormValue(heartbeatPortKey) + raftReplicaPort = r.FormValue(replicaPortKey) if mediaType, err = extractMediaType(r); err != nil { return diff --git a/master/cluster.go b/master/cluster.go index 16f67b40f..29d0d349c 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -1134,12 +1134,14 @@ func (c *Cluster) addMetaNode(nodeAddr, heartbeatPort, replicaPort, zoneName str } // compatible with old version in which raft heartbeat port and replica port did not persist - metaNode.Lock() - defer metaNode.Unlock() - metaNode.HeartbeatPort = heartbeatPort - metaNode.ReplicaPort = replicaPort - if err = c.syncUpdateMetaNode(metaNode); err != nil { - return metaNode.ID, err + if len(heartbeatPort) > 0 && len(replicaPort) > 0 { + metaNode.Lock() + defer metaNode.Unlock() + metaNode.HeartbeatPort = heartbeatPort + metaNode.ReplicaPort = replicaPort + if err = c.syncUpdateMetaNode(metaNode); err != nil { + return metaNode.ID, err + } } return metaNode.ID, nil @@ -1288,12 +1290,14 @@ func (c *Cluster) addDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zone } // compatible with old version in which raft heartbeat port and replica port did not persist - dataNode.Lock() - defer dataNode.Unlock() - dataNode.HeartbeatPort = raftHeartbeatPort - dataNode.ReplicaPort = raftReplicaPort - if err = c.syncUpdateDataNode(dataNode); err != nil { - return dataNode.ID, err + if len(raftHeartbeatPort) > 0 && len(raftReplicaPort) > 0 { + dataNode.Lock() + defer dataNode.Unlock() + dataNode.HeartbeatPort = raftHeartbeatPort + dataNode.ReplicaPort = raftReplicaPort + if err = c.syncUpdateDataNode(dataNode); err != nil { + return dataNode.ID, err + } } return dataNode.ID, nil diff --git a/master/cluster_task.go b/master/cluster_task.go index 179ef49bd..64fa8c0c2 100644 --- a/master/cluster_task.go +++ b/master/cluster_task.go @@ -1117,7 +1117,7 @@ func (c *Cluster) updateMetaNode(metaNode *MetaNode, metaPartitions []*proto.Met mp.addUpdateMetaReplicaTask(c) } - mp.updateMetaPartition(mr, metaNode) + mp.updateMetaPartition(mr, metaNode, c) vol.uidSpaceManager.pushUidMsg(mr) vol.quotaManager.quotaUpdate(mr) c.updateInodeIDUpperBound(mp, mr, threshold, metaNode) diff --git a/master/data_partition.go b/master/data_partition.go index e5f738305..07ca1e02e 100644 --- a/master/data_partition.go +++ b/master/data_partition.go @@ -770,6 +770,28 @@ func (partition *DataPartition) updateMetric(vr *proto.DataPartitionReport, data replica.DiskPath = oldDiskPath } } + + // update old partition peers, add raft ports + localPeersWithPort := make(map[string]proto.Peer) + for _, peer := range vr.LocalPeers { + if len(peer.ReplicaPort) > 0 && len(peer.HeartbeatPort) > 0 { + localPeersWithPort[peer.Addr] = peer + } + } + needUpdate := false + for i, peer := range partition.Peers { + if len(peer.ReplicaPort) == 0 || len(peer.HeartbeatPort) == 0 { + if localPeer, exist := localPeersWithPort[peer.Addr]; exist { + partition.Peers[i].ReplicaPort = localPeer.ReplicaPort + partition.Peers[i].HeartbeatPort = localPeer.HeartbeatPort + needUpdate = true + } + } + } + if needUpdate { + c.syncUpdateDataPartition(partition) + } + partition.checkAndRemoveMissReplica(dataNode.Addr) if replica.Status == proto.ReadWrite && (partition.RdOnly || replica.dataNode.RdOnly) { diff --git a/master/meta_partition.go b/master/meta_partition.go index 2fe685db0..fcabbf349 100644 --- a/master/meta_partition.go +++ b/master/meta_partition.go @@ -407,7 +407,7 @@ func (mp *MetaPartition) missingReplicaAddrs() (lackAddrs []string) { return } -func (mp *MetaPartition) updateMetaPartition(mgr *proto.MetaPartitionReport, metaNode *MetaNode) { +func (mp *MetaPartition) updateMetaPartition(mgr *proto.MetaPartitionReport, metaNode *MetaNode, c *Cluster) { if !contains(mp.Hosts, metaNode.Addr) { return } @@ -432,6 +432,27 @@ func (mp *MetaPartition) updateMetaPartition(mgr *proto.MetaPartitionReport, met mp.setStatByStorageClass() mp.setHeartBeatDone() mp.SetForbidWriteOpOfProtoVer0() + + // update old partition peers, add raft ports + localPeersWithPort := make(map[string]proto.Peer) + for _, peer := range mgr.LocalPeers { + if len(peer.ReplicaPort) > 0 && len(peer.HeartbeatPort) > 0 { + localPeersWithPort[peer.Addr] = peer + } + } + needUpdate := false + for i, peer := range mp.Peers { + if len(peer.ReplicaPort) == 0 || len(peer.HeartbeatPort) == 0 { + if localPeer, exist := localPeersWithPort[peer.Addr]; exist { + mp.Peers[i].ReplicaPort = localPeer.ReplicaPort + mp.Peers[i].HeartbeatPort = localPeer.HeartbeatPort + needUpdate = true + } + } + } + if needUpdate { + c.syncUpdateMetaPartition(mp) + } } func (mp *MetaPartition) canBeOffline(nodeAddr string, replicaNum int) (err error) { diff --git a/metanode/manager_op.go b/metanode/manager_op.go index 7c9774883..77df81580 100644 --- a/metanode/manager_op.go +++ b/metanode/manager_op.go @@ -178,6 +178,7 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet, StatByStorageClass: partition.GetStatByStorageClass(), StatByMigrateStorageClass: partition.GetMigrateStatByStorageClass(), ForbidWriteOpOfProtoVer0: mpForbidWriteVer0, + LocalPeers: mConf.Peers, } mpr.TxCnt, mpr.TxRbInoCnt, mpr.TxRbDenCnt = partition.TxGetCnt() diff --git a/metanode/partition.go b/metanode/partition.go index 23a481ed9..f26e3a63b 100644 --- a/metanode/partition.go +++ b/metanode/partition.go @@ -971,18 +971,26 @@ func (mp *metaPartition) onStop() { func (mp *metaPartition) startRaft() (err error) { var ( - heartbeatPort int - replicaPort int - peers []raftstore.PeerAddress + defaultHeartbeatPort int + defaultReplicaPort int + peers []raftstore.PeerAddress ) - //if heartbeatPort, replicaPort, err = mp.getRaftPort(); err != nil { - // return - //} + if defaultHeartbeatPort, defaultReplicaPort, err = mp.getRaftPort(); err != nil { + return + } for _, peer := range mp.config.Peers { addr := strings.Split(peer.Addr, ":")[0] - heartbeatPort, _ = strconv.Atoi(peer.HeartbeatPort) - replicaPort, _ = strconv.Atoi(peer.ReplicaPort) + + heartbeatPort, perr := strconv.Atoi(peer.HeartbeatPort) + if perr != nil { + heartbeatPort = defaultHeartbeatPort + } + replicaPort, perr := strconv.Atoi(peer.ReplicaPort) + if perr != nil { + replicaPort = defaultReplicaPort + } + rp := raftstore.PeerAddress{ Peer: raftproto.Peer{ ID: peer.ID, @@ -1016,28 +1024,28 @@ func (mp *metaPartition) stopRaft() { } } -//func (mp *metaPartition) getRaftPort() (heartbeat, replica int, err error) { -// raftConfig := mp.config.RaftStore.RaftConfig() -// heartbeatAddrSplits := strings.Split(raftConfig.HeartbeatAddr, ":") -// replicaAddrSplits := strings.Split(raftConfig.ReplicateAddr, ":") -// if len(heartbeatAddrSplits) != 2 { -// err = ErrIllegalHeartbeatAddress -// return -// } -// if len(replicaAddrSplits) != 2 { -// err = ErrIllegalReplicateAddress -// return -// } -// heartbeat, err = strconv.Atoi(heartbeatAddrSplits[1]) -// if err != nil { -// return -// } -// replica, err = strconv.Atoi(replicaAddrSplits[1]) -// if err != nil { -// return -// } -// return -//} +func (mp *metaPartition) getRaftPort() (heartbeat, replica int, err error) { + raftConfig := mp.config.RaftStore.RaftConfig() + heartbeatAddrSplits := strings.Split(raftConfig.HeartbeatAddr, ":") + replicaAddrSplits := strings.Split(raftConfig.ReplicateAddr, ":") + if len(heartbeatAddrSplits) != 2 { + err = ErrIllegalHeartbeatAddress + return + } + if len(replicaAddrSplits) != 2 { + err = ErrIllegalReplicateAddress + return + } + heartbeat, err = strconv.Atoi(heartbeatAddrSplits[1]) + if err != nil { + return + } + replica, err = strconv.Atoi(replicaAddrSplits[1]) + if err != nil { + return + } + return +} // NewMetaPartition creates a new meta partition with the specified configuration. func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaPartition { @@ -1061,9 +1069,19 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP multiVersionList: &proto.VolVersionInfoList{ TemporaryVerMap: make(map[uint64]*proto.VolVersionInfo), }, - enableAuditLog: true, } + // during upgrade process, create partition request may lack raft ports info + defaultHeartbeatPort, defaultReplicaPort, err := mp.getRaftPort() + if err == nil { + for i := range mp.config.Peers { + if len(mp.config.Peers[i].ReplicaPort) == 0 || len(mp.config.Peers[i].HeartbeatPort) == 0 { + mp.config.Peers[i].ReplicaPort = strconv.FormatInt(int64(defaultReplicaPort), 10) + mp.config.Peers[i].HeartbeatPort = strconv.FormatInt(int64(defaultHeartbeatPort), 10) + } + } + } + if manager != nil { mp.config.ForbidWriteOpOfProtoVer0 = manager.isVolForbidWriteOpOfProtoVer0(mp.config.VolName) } diff --git a/metanode/partition_fsmop.go b/metanode/partition_fsmop.go index c750e02d4..9c00ee87c 100644 --- a/metanode/partition_fsmop.go +++ b/metanode/partition_fsmop.go @@ -96,11 +96,21 @@ func (mp *metaPartition) fsmUpdatePartition(end uint64) (status uint8, func (mp *metaPartition) confAddNode(req *proto.AddMetaPartitionRaftMemberRequest, index uint64) (updated bool, err error) { var ( - heartbeatPort int - replicaPort int + defaultHeartbeatPort int + defaultreplicaPort int ) - heartbeatPort, _ = strconv.Atoi(req.AddPeer.HeartbeatPort) - replicaPort, _ = strconv.Atoi(req.AddPeer.ReplicaPort) + + if defaultHeartbeatPort, defaultreplicaPort, err = mp.getRaftPort(); err != nil { + return + } + heartbeatPort, perr := strconv.Atoi(req.AddPeer.HeartbeatPort) + if perr != nil { + heartbeatPort = defaultHeartbeatPort + } + replicaPort, perr := strconv.Atoi(req.AddPeer.ReplicaPort) + if perr != nil { + replicaPort = defaultreplicaPort + } addPeer := false for _, peer := range mp.config.Peers { diff --git a/proto/admin_proto.go b/proto/admin_proto.go index 989d9d23f..57551beaf 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -880,6 +880,7 @@ type MetaPartitionReport struct { QuotaReportInfos []*QuotaReportInfo StatByStorageClass []*StatOfStorageClass StatByMigrateStorageClass []*StatOfStorageClass + LocalPeers []Peer } // MetaNodeHeartbeatResponse defines the response to the meta node heartbeat request.