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

feat(master): add param to limit replica location;once raft use diverse port, can't be changed

Browse Source 
Signed-off-by: morphes1995 <morphes1995@gmail.com>
This commit is contained in:
morphes1995 2024-12-02 15:51:03 +08:00 committed by leonrayang
parent 9a334cd7e8
commit bc08a804cb
10 changed files with 232 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
docs-zh/source/ops/configs/master.md
View File

@ -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 |
## 配置示例

1
docs/source/ops/configs/master.md
View File

@ -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

16
master/api_service.go
View File

@ -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

32
master/cluster.go
View File

@ -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
}

12
master/cluster_task.go
View File

@ -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
}

83
master/config.go
View File

@ -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
}

15
master/data_partition.go
View File

@ -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 {

26
master/node_selector.go
View File

@ -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)

5
master/server.go
View File

@ -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) {

4
master/vol.go
View File

@ -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)