From bc08a804cb0d752cbec19cd51eb59b16e81d6fb2 Mon Sep 17 00:00:00 2001 From: morphes1995 Date: Mon, 2 Dec 2024 15:51:03 +0800 Subject: [PATCH] feat(master): add param to limit replica location;once raft use diverse port, can't be changed Signed-off-by: morphes1995 --- docs-zh/source/ops/configs/master.md | 79 +++++++++++++------------- docs/source/ops/configs/master.md | 1 + master/api_service.go | 16 ++++++ master/cluster.go | 32 +++++++++++ master/cluster_task.go | 12 ++++ master/config.go | 83 ++++++++++++++++++++++++++++ master/data_partition.go | 15 +++++ master/node_selector.go | 26 ++++++++- master/server.go | 5 ++ master/vol.go | 4 ++ 10 files changed, 232 insertions(+), 41 deletions(-) diff --git a/docs-zh/source/ops/configs/master.md b/docs-zh/source/ops/configs/master.md index c27ff6289..d727b16ce 100755 --- a/docs-zh/source/ops/configs/master.md +++ b/docs-zh/source/ops/configs/master.md @@ -3,45 +3,46 @@ CubeFS 使用 **JSON** 作为配置文件的格式. -| 配置项 | 类型 | 描述 | 必需 | 默认值 | -| :---------------------------------- | :----- |:--------------------------------------------------------------------------------------------------------------|:----| :--------- | -| role | string | 进程的角色，值只能是 master | 是 | | -| ip | string | 主机ip | 是 | | -| listen | string | http服务监听的端口号 | 是 | | -| prof | string | golang pprof 端口号 | 是 | | -| id | string | 区分不同的master节点 | 是 | | -| peers | string | raft复制组成员信息 | 是 | | -| logDir | string | 日志文件存储目录 | 是 | | -| logLevel | string | 日志级别 | 否 | error | -| retainLogs | string | 保留多少条raft日志. | 是 | | -| walDir | string | raft wal日志存储目录. | 是 | | -| storeDir | string | RocksDB数据存储目录.此目录必须存在，如果目录不存在，无法启动服务 | 是 | | -| clusterName | string | 集群名字 | 是 | | -| ebsAddr | string | 纠删码子系统的地址，使用纠删码子系统时需配置 | 否 | | -| exporterPort | int | prometheus获取监控数据端口 | 否 | | -| consulAddr | string | consul注册地址，供prometheus exporter使用 | 否 | | -| metaNodeReservedMem | string | 元数据节点预留内存大小，单位：字节 | 否 | 1073741824 | -| heartbeatPort | string | raft心跳通信端口 | 否 | 5901 | -| replicaPort | string | raft数据传输端口 | 否 | 5902 | -| nodeSetCap | string | NodeSet的容量 | 否 | 18 | -| missingDataPartitionInterval | string | 当此时间段内没有收到副本的心跳，该副本被认为已丢失，单位：s | 否 | 24h | -| dataPartitionTimeOutSec | string | 当此时间段内没有收到副本的心跳，该副本被认为非存活，单位：s | 否 | 10min | -| numberOfDataPartitionsToLoad | string | 一次最多检查多少数据分片 | 否 | 40 | -| secondsToFreeDataPartitionAfterLoad | string | 在多少秒之后开始释放由加载数据分片任务占用的内存 | 否 | 300 | -| tickInterval | string | 检查心跳和选举超时的计时器间隔，单位：ms | 否 | 500 | -| electionTick | string | 在计时器重置多少次时，选举超时 | 否 | 5 | -| bindIp | bool | 是否仅在主机ip上监听连接 | 否 | false | -| faultDomain | bool | 是否启用故障域 | 否 | false | -| faultDomainBuildAsPossible | bool | 若可用的故障域数量少于预期的故障域数量，是否仍尽可能地去构建nodeSetGroup | 否 | false | -| faultDomainGrpBatchCnt | string | 可用的故障域数量 | 否 | 3 | -| dpNoLeaderReportIntervalSec | string | 数据分片没有leader时，多久上报一次，单位：s | 否 | 60 | -| mpNoLeaderReportIntervalSec | string | 元数据分片没有leader时，多久上报一次，单位：s | 否 | 60 | -| maxQuotaNumPerVol | string | 单个卷最大的配额数 | 否 | 100 | -| volForceDeletion | bool | 非空的卷是否可以删除 | 否 | true | -| volDeletionDentryThreshold | int | 如果非空的卷不可以直接删除， 该参数定义了一个阈值，只有一个卷的dentry个数小于等于该阈值时才可以被删除 | 否 | 0 | -| enableLogPanicHook | bool | (实验性) Hook `panic` 函数以便在执行`panic`之前使日志落盘 | No | false | -| enableDirectDeleteVol | bool | 用于控制是否直接删除卷，`true` 将会直接删除，`false` 延迟删除 | No | true | -| raftPartitionCanUsingDifferentPort | bool | 数据/元数据分区是否可以使用不同的raft heartbeatPort 和 replicatePort， 如果可以，我们可以在一台机器上部署多个datanode/metanode进程 | 否 | false | +| 配置项 | 类型 | 描述 | 必需 | 默认值 | +| :---------------------------------- | :----- |:-------------------------------------------------------------------------------------------------------------|:--------| :--------- | +| role | string | 进程的角色，值只能是 master | 是 | | +| ip | string | 主机ip | 是 | | +| listen | string | http服务监听的端口号 | 是 | | +| prof | string | golang pprof 端口号 | 是 | | +| id | string | 区分不同的master节点 | 是 | | +| peers | string | raft复制组成员信息 | 是 | | +| logDir | string | 日志文件存储目录 | 是 | | +| logLevel | string | 日志级别 | 否 | error | +| retainLogs | string | 保留多少条raft日志. | 是 | | +| walDir | string | raft wal日志存储目录. | 是 | | +| storeDir | string | RocksDB数据存储目录.此目录必须存在，如果目录不存在，无法启动服务 | 是 | | +| clusterName | string | 集群名字 | 是 | | +| ebsAddr | string | 纠删码子系统的地址，使用纠删码子系统时需配置 | 否 | | +| exporterPort | int | prometheus获取监控数据端口 | 否 | | +| consulAddr | string | consul注册地址，供prometheus exporter使用 | 否 | | +| metaNodeReservedMem | string | 元数据节点预留内存大小，单位：字节 | 否 | 1073741824 | +| heartbeatPort | string | raft心跳通信端口 | 否 | 5901 | +| replicaPort | string | raft数据传输端口 | 否 | 5902 | +| nodeSetCap | string | NodeSet的容量 | 否 | 18 | +| missingDataPartitionInterval | string | 当此时间段内没有收到副本的心跳，该副本被认为已丢失，单位：s | 否 | 24h | +| dataPartitionTimeOutSec | string | 当此时间段内没有收到副本的心跳，该副本被认为非存活，单位：s | 否 | 10min | +| numberOfDataPartitionsToLoad | string | 一次最多检查多少数据分片 | 否 | 40 | +| secondsToFreeDataPartitionAfterLoad | string | 在多少秒之后开始释放由加载数据分片任务占用的内存 | 否 | 300 | +| tickInterval | string | 检查心跳和选举超时的计时器间隔，单位：ms | 否 | 500 | +| electionTick | string | 在计时器重置多少次时，选举超时 | 否 | 5 | +| bindIp | bool | 是否仅在主机ip上监听连接 | 否 | false | +| faultDomain | bool | 是否启用故障域 | 否 | false | +| faultDomainBuildAsPossible | bool | 若可用的故障域数量少于预期的故障域数量，是否仍尽可能地去构建nodeSetGroup | 否 | false | +| faultDomainGrpBatchCnt | string | 可用的故障域数量 | 否 | 3 | +| dpNoLeaderReportIntervalSec | string | 数据分片没有leader时，多久上报一次，单位：s | 否 | 60 | +| mpNoLeaderReportIntervalSec | string | 元数据分片没有leader时，多久上报一次，单位：s | 否 | 60 | +| maxQuotaNumPerVol | string | 单个卷最大的配额数 | 否 | 100 | +| volForceDeletion | bool | 非空的卷是否可以删除 | 否 | true | +| volDeletionDentryThreshold | int | 如果非空的卷不可以直接删除， 该参数定义了一个阈值，只有一个卷的dentry个数小于等于该阈值时才可以被删除 | 否 | 0 | +| enableLogPanicHook | bool | (实验性) Hook `panic` 函数以便在执行`panic`之前使日志落盘 | No | false | +| enableDirectDeleteVol | bool | 用于控制是否直接删除卷，`true` 将会直接删除，`false` 延迟删除 | No | true | +| raftPartitionCanUsingDifferentPort | bool | 数据/元数据分区是否可以使用不同的raft heartbeatPort 和 replicatePort， 如果可以，我们可以在一台机器上部署多个datanode/metanode进程 | 否 | false | +| allowMultipleReplicasOnSameMachine | bool | 数据分区/元数据分区的副本是否允许在同一台机器上 | 否 | true | ## 配置示例 diff --git a/docs/source/ops/configs/master.md b/docs/source/ops/configs/master.md index 7cdf88a60..0fdd6bfb4 100755 --- a/docs/source/ops/configs/master.md +++ b/docs/source/ops/configs/master.md @@ -42,6 +42,7 @@ CubeFS uses **JSON** as the format of the configuration file. | enableLogPanicHook | bool | (Experimental) Hook `panic` function to flush log before executing `panic` | No | false | | enableDirectDeleteVol | bool | to control the support for delayed volume deletion. `true``, will delete volume directly | No | true | | raftPartitionCanUsingDifferentPort | bool | whether data partition/meta partition can use different raft heartbeatPort and replicatePort. if so we can deploy multiple datanode/metanode on single machine | No | false | +| allowMultipleReplicasOnSameMachine | bool | whether replicas of data partition/meta partition can locate on same machine | No | true | ## Configuration Example diff --git a/master/api_service.go b/master/api_service.go index 895a687de..611a3e70f 100644 --- a/master/api_service.go +++ b/master/api_service.go @@ -1795,6 +1795,13 @@ func (m *Server) addDataReplica(w http.ResponseWriter, r *http.Request) { return } + dp.RLock() + newHosts := append(dp.Hosts, addr) + dp.RUnlock() + if err = m.cluster.checkMultipleReplicasOnSameMachine(newHosts); err != nil { + return + } + retry := 0 for { if !dp.setRestoreReplicaForbidden() { @@ -1890,6 +1897,7 @@ func (m *Server) addMetaReplica(w http.ResponseWriter, r *http.Request) { addr string mp *MetaPartition partitionID uint64 + allHosts []string err error ) metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminAddMetaReplica)) @@ -1907,6 +1915,14 @@ func (m *Server) addMetaReplica(w http.ResponseWriter, r *http.Request) { return } + mp.RLock() + allHosts = append(mp.Hosts, addr) + mp.RUnlock() + if err = m.cluster.checkMultipleReplicasOnSameMachine(allHosts); err != nil { + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()}) + return + } + if err = m.cluster.addMetaReplica(mp, addr); err != nil { sendErrReply(w, r, newErrHTTPReply(err)) return diff --git a/master/cluster.go b/master/cluster.go index 8bbc2c508..728677871 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -18,6 +18,7 @@ import ( "encoding/json" "fmt" "math" + "net" "net/http" "sort" "strconv" @@ -1851,6 +1852,9 @@ func (c *Cluster) createDataPartition(volName string, preload *DataPartitionPreL goto errHandler } } + if err = c.checkMultipleReplicasOnSameMachine(targetHosts); err != nil { + goto errHandler + } if partitionID, err = c.idAlloc.allocateDataPartitionID(); err != nil { goto errHandler @@ -2720,6 +2724,7 @@ ERR: func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataPartition, raftForce bool, errMsg string) (err error) { var ( targetHosts []string + finalHosts []string newAddr string msg string dataNode *DataNode @@ -2775,6 +2780,19 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti goto errHandler } + dp.RLock() + finalHosts = append(dp.Hosts, newAddr) // add new one + dp.RUnlock() + for i, host := range finalHosts { + if host == srcAddr { + finalHosts = append(finalHosts[:i], finalHosts[i+1:]...) // remove old one + break + } + } + if err = c.checkMultipleReplicasOnSameMachine(finalHosts); err != nil { + goto errHandler + } + if targetAddr != "" { targetHosts = []string{targetAddr} if err = c.checkDataNodesMediaTypeForMigrate(dataNode, targetAddr); err != nil { @@ -6163,3 +6181,17 @@ func (c *Cluster) getVolOpLog(volName string) proto.OpLogView { } return opv } + +func (c *Cluster) checkMultipleReplicasOnSameMachine(hosts []string) (err error) { + if !c.cfg.AllowMultipleReplicasOnSameMachine { + distinctIp := map[string]struct{}{} + for _, hostStr := range hosts { + ip, _, _ := net.SplitHostPort(hostStr) + if _, exist := distinctIp[ip]; exist { + return fmt.Errorf("Don't allow multiple replicas on same machine while create dp/mp. Multiple replicas locate on [%v] ", ip) + } + distinctIp[ip] = struct{}{} + } + } + return nil +} diff --git a/master/cluster_task.go b/master/cluster_task.go index 64fa8c0c2..f7bced905 100644 --- a/master/cluster_task.go +++ b/master/cluster_task.go @@ -102,6 +102,7 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti zone *Zone ns *nodeSet excludeNodeSets []uint64 + finalHosts []string oldHosts []string zones []string ) @@ -167,6 +168,17 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti } } + finalHosts = append(oldHosts, newPeers[0].Addr) // add new one + for i, host := range finalHosts { + if host == srcAddr { + finalHosts = append(finalHosts[:i], finalHosts[i+1:]...) // remove old one + break + } + } + if err = c.checkMultipleReplicasOnSameMachine(finalHosts); err != nil { + return err + } + if err = c.deleteMetaReplica(mp, srcAddr, false, false); err != nil { goto errHandler } diff --git a/master/config.go b/master/config.go index c465c213f..8be4296ce 100644 --- a/master/config.go +++ b/master/config.go @@ -15,8 +15,11 @@ package master import ( + "encoding/json" "fmt" syslog "log" + "os" + "path" "strconv" "strings" @@ -61,6 +64,7 @@ const ( cfgLegacyDataMediaType = "legacyDataMediaType" // for hybrid cloud upgrade cfgRaftPartitionCanUsingDifferentPort = "raftPartitionCanUsingDifferentPort" + cfgAllowMultipleReplicasOnSameMachine = "allowMultipleReplicasOnSameMachine" ) // default value @@ -171,6 +175,14 @@ type clusterConfig struct { forbidWriteOpOfProtoVer0 bool raftPartitionCanUsingDifferentPort bool // whether data partition/meta partition can use different raft heartbeat port and replicate port. if so we can deploy multiple datanode/metanode on single machine + AllowMultipleReplicasOnSameMachine bool // whether dp/mp replicas can locate on same machine, default true +} +type clusterConstConfig struct { + RaftPartitionCanUsingDifferentPort bool `json:"raftPartitionCanUsingDifferentPort"` +} + +func (c *clusterConstConfig) Equals(c2 *clusterConstConfig) bool { + return c.RaftPartitionCanUsingDifferentPort == c2.RaftPartitionCanUsingDifferentPort } func newClusterConfig() (cfg *clusterConfig) { @@ -233,3 +245,74 @@ func (cfg *clusterConfig) parsePeers(peerStr string) error { } return nil } + +func (cfg *clusterConfig) CheckOrStoreConstCfg(fileDir, fileName string) (err error) { + constCfg := &clusterConstConfig{ + RaftPartitionCanUsingDifferentPort: cfg.raftPartitionCanUsingDifferentPort, + } + + filePath := path.Join(fileDir, fileName) + var buf []byte + buf, err = os.ReadFile(filePath) + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("read config file %v failed: %v", filePath, err) + } + + if os.IsNotExist(err) || len(buf) == 0 { + // Persist configuration to disk + if buf, err = json.Marshal(constCfg); err != nil { + return fmt.Errorf("marshal const config failed: %v", err) + } + if err = os.MkdirAll(fileDir, 0o755); err != nil { + return fmt.Errorf("make directory %v filed: %v", fileDir, err) + } + var file *os.File + if file, err = os.OpenFile(filePath, os.O_CREATE|os.O_RDWR, 0o755); err != nil { + return fmt.Errorf("create config file %v failed: %v", filePath, err) + } + defer func() { + _ = file.Close() + if err != nil { + _ = os.Remove(filePath) + } + }() + if _, err = file.Write(buf); err != nil { + return fmt.Errorf("write config file %v failed: %v", filePath, err) + } + if err = file.Sync(); err != nil { + return fmt.Errorf("sync config file %v failed: %v", filePath, err) + } + return nil + } + // Load and check stored const configuration + storedConstCfg := new(clusterConstConfig) + if err = json.Unmarshal(buf, storedConstCfg); err != nil { + return fmt.Errorf("unmarshal master const config %v failed: %v", filePath, err) + } + + if storedConstCfg.RaftPartitionCanUsingDifferentPort && !constCfg.RaftPartitionCanUsingDifferentPort { + return fmt.Errorf("Param raftPartitionCanUsingDifferentPort once enabled, can not be changed anymore ") + } + + if buf, err = json.Marshal(constCfg); err != nil { + return fmt.Errorf("marshal const config failed: %v", err) + } + var file *os.File + if file, err = os.OpenFile(filePath, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0o755); err != nil { + return fmt.Errorf("create config file %v failed: %v", filePath, err) + } + defer func() { + _ = file.Close() + if err != nil { + _ = os.Remove(filePath) + } + }() + if _, err = file.Write(buf); err != nil { + return fmt.Errorf("write config file %v failed: %v", filePath, err) + } + if err = file.Sync(); err != nil { + return fmt.Errorf("sync config file %v failed: %v", filePath, err) + } + + return nil +} diff --git a/master/data_partition.go b/master/data_partition.go index d9bc1bcc8..3a6f9d529 100644 --- a/master/data_partition.go +++ b/master/data_partition.go @@ -1389,6 +1389,7 @@ func (partition *DataPartition) Decommission(c *Cluster) bool { srcReplica *DataReplica resetDecommissionDst = true begin = time.Now() + finalHosts []string ) if partition.GetDecommissionStatus() == DecommissionInitial { @@ -1404,6 +1405,20 @@ func (partition *DataPartition) Decommission(c *Cluster) bool { partition.markRollbackFailed(false) return false } + + partition.RLock() + finalHosts = append(partition.Hosts, targetAddr) // add new one + partition.RUnlock() + for i, host := range finalHosts { + if host == srcAddr { + finalHosts = append(finalHosts[:i], finalHosts[i+1:]...) // remove old one + break + } + } + if err = c.checkMultipleReplicasOnSameMachine(finalHosts); err != nil { + goto errHandler + } + partition.SetDecommissionStatus(DecommissionPrepare) err = c.syncUpdateDataPartition(partition) if err != nil { diff --git a/master/node_selector.go b/master/node_selector.go index bc610f20c..21e1e8209 100644 --- a/master/node_selector.go +++ b/master/node_selector.go @@ -305,10 +305,22 @@ func (s *AvailableSpaceFirstNodeSelector) Select(ns *nodeSet, excludeHosts []str return s.getNodeAvailableSpace(sortedNodes[i]) > s.getNodeAvailableSpace(sortedNodes[j]) }) + // select replica number of nodes, try to avoid multiple replicas of a partition locate on same machine + // If raftPartitionCanUsingDifferentPort is enabled, sortedNodes may contain nodes with same ip + + // Consider the below case： + // Machine1 has 3 dn process (IP1:17310, IP1:18310, IP1:19310) + // Machine2 has 2 dn process (IP2:17310, IP2:18310) + // Machine3 has 1 dn process (IP3:17310) + // 1) When create 3 replicas dp, we will select IP1:17310, IP2:17310, IP3:17310, inner loop only be executed once + // 2) When create 5 replicas dp, we will select IP1:17310, IP2:17310, IP3:17310,IP1:18310, IP2:18310 + // inner loop be executed twice, first loop select out IP1:17310, IP2:17310, IP3:17310 + // because nodes with distinct ip can’t satisfy the replica requirement, need second loop , and select out IP1:18310, IP2:18310 excludedNodes := make([]Node, 0) distinctIpSet := make(map[string]struct{}) + // outer loop: select until we get replica number of nodes for len(orderHosts) < replicaNum { - // select set of nodes with distinct ip in each iteration, try to avoid multiple replicas of a partition locate on same machine + // for each execution of inner loop, select nodes with distinct ip, try to avoid multiple replicas of a partition locate on same machine for i := 0; i < len(sortedNodes); i++ { node := sortedNodes[i] addr := node.GetAddr() @@ -329,6 +341,8 @@ func (s *AvailableSpaceFirstNodeSelector) Select(ns *nodeSet, excludeHosts []str break } } + + // number of nodes with distinct ip can not satisfy replica requirement sortedNodes = excludedNodes distinctIpSet = make(map[string]struct{}) } @@ -452,6 +466,7 @@ func (s *StrawNodeSelector) getWeight(node Node) float64 { return float64(node.GetAvailableSpace()) / util.GB } +// select a node with max straw and it's ip didn't exist in excludedIpSet func (s *StrawNodeSelector) selectOneNode(nodes []Node, excludedIpSet map[string]struct{}) (index int, maxNode Node) { maxStraw := float64(0) maxStrawNodeIp := "" @@ -503,7 +518,13 @@ func (s *StrawNodeSelector) Select(ns *nodeSet, excludeHosts []string, replicaNu distinctIpSet := make(map[string]struct{}) orderHosts := make([]string, 0) + + // select replica number of nodes, try to avoid multiple replicas of a partition locate on same machine + // If raftPartitionCanUsingDifferentPort is enabled, candidate nodes may contain nodes with same ip + + // outer loop: select until we get replica number of nodes for len(orderHosts) < replicaNum { + // for each execution of inner loop, select nodes with distinct ip, try to avoid multiple replicas of a partition locate on same machine for { index, node := s.selectOneNode(nodes, distinctIpSet) if index == -1 { @@ -523,6 +544,7 @@ func (s *StrawNodeSelector) Select(ns *nodeSet, excludeHosts []string, replicaNu break } } + // number of nodes with distinct ip can not satisfy replica requirement distinctIpSet = make(map[string]struct{}) } @@ -566,7 +588,7 @@ func NewNodeSelector(name string, nodeType NodeType) NodeSelector { func (ns *nodeSet) getAvailMetaNodeHosts(excludeHosts []string, replicaNum int) (newHosts []string, peers []proto.Peer, err error) { ns.nodeSelectLock.Lock() defer ns.nodeSelectLock.Unlock() - // we need a read lock to block the modify of node selector + // we need a read lock to block the modification of node selector ns.metaNodeSelectorLock.RLock() defer ns.metaNodeSelectorLock.RUnlock() return ns.metaNodeSelector.Select(ns, excludeHosts, replicaNum) diff --git a/master/server.go b/master/server.go index a3c4f0a98..cc7004405 100644 --- a/master/server.go +++ b/master/server.go @@ -398,6 +398,11 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) { enableDirectDeleteVol = cfg.GetBoolWithDefault(cfgEnableDirectDeleteVol, true) m.config.raftPartitionCanUsingDifferentPort = cfg.GetBoolWithDefault(cfgRaftPartitionCanUsingDifferentPort, false) + m.config.AllowMultipleReplicasOnSameMachine = cfg.GetBoolWithDefault(cfgAllowMultipleReplicasOnSameMachine, true) + + if err = m.config.CheckOrStoreConstCfg(m.storeDir, config.DefaultConstConfigFile); err != nil { + return + } m.config.cfgDataMediaType = uint32(cfg.GetInt64(cfgLegacyDataMediaType)) if m.config.cfgDataMediaType != 0 && !proto.IsValidMediaType(m.config.cfgDataMediaType) { diff --git a/master/vol.go b/master/vol.go index 6193cc374..4db4e1be5 100644 --- a/master/vol.go +++ b/master/vol.go @@ -1795,6 +1795,10 @@ func (vol *Vol) doCreateMetaPartition(c *Cluster, start, end uint64) (mp *MetaPa } } + if err = c.checkMultipleReplicasOnSameMachine(hosts); err != nil { + return nil, err + } + log.LogInfof("target meta hosts:%v,peers:%v", hosts, peers) if partitionID, err = c.idAlloc.allocateMetaPartitionID(); err != nil { return nil, errors.NewError(err)