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): conf persist in rocksdb; if raft use diverse port, old version nodes can't register

Browse Source 
Signed-off-by: morphes1995 <morphes1995@gmail.com>
This commit is contained in:
morphes1995 2024-12-15 22:55:17 +08:00 committed by leonrayang
parent ab398d669d
commit fb4f541413
11 changed files with 168 additions and 173 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,8 +3,8 @@
CubeFS 使用 **JSON** 作为配置文件的格式.
| 配置项 | 类型 | 描述 | 必需 | 默认值 |
| :---------------------------------- | :----- |:-------------------------------------------------------------------------------------------------------------|:--------| :--------- |
| 配置项 | 类型 | 描述 | 必需 | 默认值 |
|:------------------------------------| :----- |:-------------------------------------------------------------------------------------------------------------|:--------| :--------- |
| role | string | 进程的角色，值只能是 master | 是 | |
| ip | string | 主机ip | 是 | |
| listen | string | http服务监听的端口号 | 是 | |
@ -41,7 +41,7 @@ CubeFS 使用 **JSON** 作为配置文件的格式.
| 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 |
| raftPartitionCanUseDifferentPort | bool | 数据/元数据分区是否可以使用不同的raft heartbeatPort 和 replicatePort， 如果可以，我们可以在一台机器上部署多个datanode/metanode进程 | 否 | false |
| allowMultipleReplicasOnSameMachine | bool | 数据分区/元数据分区的副本是否允许在同一台机器上 | 否 | true |
## 配置示例

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

@ -41,7 +41,7 @@ CubeFS uses **JSON** as the format of the configuration file.
| volDeletionDentryThreshold | int | if the non-empty volume can't be deleted directly , this param define a threshold , only volumes with a dentry count that is less than or equal to the threshold can be deleted | No | 0 |
| 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 |
| raftPartitionCanUseDifferentPort | 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

46
master/cluster.go
View File

@ -1123,14 +1123,19 @@ func (c *Cluster) updateMetaNodeBaseInfo(nodeAddr string, id uint64) (err error)
return
}
// RaftPartitionCanUsingDifferentPortEnabled check whether raftPartitionCanUsingDifferentPort param become effective
// raftPartitionCanUsingDifferentPort param take into force only when
// 1. raftPartitionCanUsingDifferentPort set true,
// 2. all data nodes and meta nodes are registered with HeartbeatPort and ReplicaPort
// RaftPartitionCanUsingDifferentPortEnabled check whether raft partition can use different port or not
func (c *Cluster) RaftPartitionCanUsingDifferentPortEnabled() bool {
if !c.cfg.raftPartitionCanUsingDifferentPort {
if c.cfg.raftPartitionAlreadyUseDifferentPort.Load() {
// this cluster has already enabled this feature
return true
}
if !c.cfg.raftPartitionCanUseDifferentPort.Load() {
// user currently don't enable this feature
return false
}
// user currently try to enable this feature
enabled := true
c.mnMutex.RLock()
c.metaNodes.Range(func(addr, node interface{}) bool {
@ -1157,6 +1162,19 @@ func (c *Cluster) RaftPartitionCanUsingDifferentPortEnabled() bool {
c.dnMutex.RUnlock()
}
if enabled && !c.cfg.raftPartitionAlreadyUseDifferentPort.Load() {
// all data nodes and meta nodes are registered with HeartbeatPort and ReplicaPort
// this feature now is enabled, we update cluster cfg and store
c.cfg.raftPartitionAlreadyUseDifferentPort.Store(true)
if err := c.syncPutCluster(); err != nil {
log.LogErrorf("error syncPutCluster when set raftPartitionAlreadyUseDifferentPort to true, err:%v", err)
c.cfg.raftPartitionAlreadyUseDifferentPort.Store(false) // set back to false, let syncPutCluster try again in future
return false
}
log.LogInfof("all data nodes and meta nodes are registered with HeartbeatPort and ReplicaPort, " +
"raft partition use different port feature now is enabled")
}
return enabled
}
@ -1171,11 +1189,11 @@ func (c *Cluster) addMetaNode(nodeAddr, heartbeatPort, replicaPort, zoneName str
return metaNode.ID, fmt.Errorf("addr already in nodeset [%v]", nodeAddr)
}
// compatible with old version in which raft heartbeat port and replica port did not persist
if len(heartbeatPort) > 0 && len(replicaPort) > 0 {
metaNode.Lock()
defer metaNode.Unlock()
if len(metaNode.HeartbeatPort) == 0 || len(metaNode.ReplicaPort) == 0 {
// compatible with old version in which raft heartbeat port and replica port did not persist
metaNode.HeartbeatPort = heartbeatPort
metaNode.ReplicaPort = replicaPort
if err = c.syncUpdateMetaNode(metaNode); err != nil {
@ -1186,6 +1204,13 @@ func (c *Cluster) addMetaNode(nodeAddr, heartbeatPort, replicaPort, zoneName str
return metaNode.ID, nil
}
if c.cfg.raftPartitionCanUseDifferentPort.Load() {
if len(heartbeatPort) == 0 || len(replicaPort) == 0 {
err = fmt.Errorf("when master enable raftPartitionCanUseDifferentPort, only allow new metanode with valid heartbeatPort and replicaPort to register. "+
"metanode(%v, heartbeatPort:%v, replicaPort:%v) may need to upgrade", nodeAddr, heartbeatPort, replicaPort)
return
}
}
metaNode = newMetaNode(nodeAddr, heartbeatPort, replicaPort, zoneName, c.Name)
metaNode.MpCntLimit = newLimitCounter(&c.cfg.MaxMpCntLimit, defaultMaxMpCntLimit)
@ -1329,11 +1354,11 @@ func (c *Cluster) addDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zone
return dataNode.ID, fmt.Errorf("mediaType not equalt old, new %v, old %v", mediaType, dataNode.MediaType)
}
// compatible with old version in which raft heartbeat port and replica port did not persist
if len(raftHeartbeatPort) > 0 && len(raftReplicaPort) > 0 {
dataNode.Lock()
defer dataNode.Unlock()
if len(dataNode.HeartbeatPort) == 0 || len(dataNode.ReplicaPort) == 0 {
// compatible with old version in which raft heartbeat port and replica port did not persist
dataNode.HeartbeatPort = raftHeartbeatPort
dataNode.ReplicaPort = raftReplicaPort
if err = c.syncUpdateDataNode(dataNode); err != nil {
@ -1344,6 +1369,13 @@ func (c *Cluster) addDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zone
return dataNode.ID, nil
}
if c.cfg.raftPartitionCanUseDifferentPort.Load() {
if len(raftHeartbeatPort) == 0 || len(raftReplicaPort) == 0 {
err = fmt.Errorf("when master enable raftPartitionCanUseDifferentPort, only allow new datanode with valid heartbeatPort and replicaPort to register. "+
"datanode(%v, heartbeatPort:%v, replicaPort:%v) may need to upgrade", nodeAddr, raftHeartbeatPort, raftReplicaPort)
return
}
}
needPersistZone := false
dataNode = newDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName, c.Name, mediaType)

95
master/config.go
View File

@ -18,11 +18,12 @@ import (
"encoding/json"
"fmt"
syslog "log"
"os"
"path"
"strconv"
"strings"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
pt "github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
@ -63,7 +64,7 @@ const (
cfgLegacyDataMediaType = "legacyDataMediaType" // for hybrid cloud upgrade
cfgRaftPartitionCanUsingDifferentPort = "raftPartitionCanUsingDifferentPort"
cfgRaftPartitionCanUseDifferentPort = "raftPartitionCanUseDifferentPort"
cfgAllowMultipleReplicasOnSameMachine = "allowMultipleReplicasOnSameMachine"
)
@ -174,15 +175,12 @@ type clusterConfig struct {
// PacketProtoVersion-1: from hybrid cloud version
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"`
}
// 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
raftPartitionCanUseDifferentPort atomicutil.Bool
raftPartitionAlreadyUseDifferentPort atomicutil.Bool
func (c *clusterConstConfig) Equals(c2 *clusterConstConfig) bool {
return c.RaftPartitionCanUsingDifferentPort == c2.RaftPartitionCanUsingDifferentPort
AllowMultipleReplicasOnSameMachine bool // whether dp/mp replicas can locate on same machine, default true
}
func newClusterConfig() (cfg *clusterConfig) {
@ -246,68 +244,31 @@ func (cfg *clusterConfig) parsePeers(peerStr string) error {
return nil
}
func (cfg *clusterConfig) CheckOrStoreConstCfg(fileDir, fileName string) (err error) {
constCfg := &clusterConstConfig{
RaftPartitionCanUsingDifferentPort: cfg.raftPartitionCanUsingDifferentPort,
func (cfg *clusterConfig) checkRaftPartitionCanUseDifferentPort(m *Server, optVal bool) (err error) {
cfg.raftPartitionCanUseDifferentPort.Store(optVal)
clusterCfg, err := m.rocksDBStore.SeekForPrefix([]byte(clusterPrefix))
if err != nil {
err = fmt.Errorf("action[checkConfig] error when load cluster config form rocksdb,err:%v", err.Error())
return err
}
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)
for _, c := range clusterCfg {
cv := &clusterValue{}
if err = json.Unmarshal(c, cv); err != nil {
log.LogErrorf("action[checkRaftPartitionCanUseDifferentPort], unmarshal err:%v", err.Error())
return err
}
if err = cfg.saveFile(filePath, buf); err != nil {
return
}
}
// 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 cv.Name != m.clusterName {
log.LogErrorf("action[checkRaftPartitionCanUseDifferentPort] loaded cluster value: %+v", cv)
continue
}
if !constCfg.Equals(storedConstCfg) {
if buf, err = json.Marshal(constCfg); err != nil {
return fmt.Errorf("marshal const config failed: %v", err)
}
if err = cfg.saveFile(filePath, buf); err != nil {
return
if cv.RaftPartitionAlreadyUseDifferentPort && !cfg.raftPartitionCanUseDifferentPort.Load() {
// raft partition has already use different port, but user want to disable this param again
return fmt.Errorf("raft partition has already use different port, can not try to disable this param again")
}
cfg.raftPartitionAlreadyUseDifferentPort.Store(cv.RaftPartitionAlreadyUseDifferentPort)
}
return nil
}
func (cfg *clusterConfig) saveFile(path string, buf []byte) (err error) {
if err = os.MkdirAll(path, 0o755); err != nil {
return fmt.Errorf("make directory %v filed: %v", path, err)
}
var file *os.File
if file, err = os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o755); err != nil {
return fmt.Errorf("create config file %v failed: %v", path, err)
}
defer func() {
_ = file.Close()
if err != nil {
_ = os.Remove(path)
}
}()
if _, err = file.Write(buf); err != nil {
return fmt.Errorf("write config file %v failed: %v", path, err)
}
if err = file.Sync(); err != nil {
return fmt.Errorf("sync config file %v failed: %v", path, err)
}
return nil
}

6
master/data_partition.go
View File

@ -774,10 +774,10 @@ func (partition *DataPartition) updateMetric(vr *proto.DataPartitionReport, data
if c.RaftPartitionCanUsingDifferentPortEnabled() {
// update old partition peers, add raft ports
localPeers := make(map[string]proto.Peer)
for i, peer := range vr.LocalPeers {
for _, peer := range vr.LocalPeers {
if len(peer.ReplicaPort) == 0 || len(peer.HeartbeatPort) == 0 {
vr.LocalPeers[i].ReplicaPort = dataNode.ReplicaPort
vr.LocalPeers[i].HeartbeatPort = dataNode.HeartbeatPort
peer.ReplicaPort = dataNode.ReplicaPort
peer.HeartbeatPort = dataNode.HeartbeatPort
}
localPeers[peer.Addr] = peer
}

7
master/meta_partition.go
View File

@ -436,10 +436,10 @@ func (mp *MetaPartition) updateMetaPartition(mgr *proto.MetaPartitionReport, met
if c.RaftPartitionCanUsingDifferentPortEnabled() {
// update old partition peers, add raft ports
localPeers := map[string]proto.Peer{}
for i, peer := range mgr.LocalPeers {
for _, peer := range mgr.LocalPeers {
if len(peer.ReplicaPort) == 0 || len(peer.HeartbeatPort) == 0 {
mgr.LocalPeers[i].ReplicaPort = metaNode.ReplicaPort
mgr.LocalPeers[i].HeartbeatPort = metaNode.HeartbeatPort
peer.ReplicaPort = metaNode.ReplicaPort
peer.HeartbeatPort = metaNode.HeartbeatPort
}
localPeers[peer.Addr] = peer
}
@ -457,7 +457,6 @@ func (mp *MetaPartition) updateMetaPartition(mgr *proto.MetaPartitionReport, met
c.syncUpdateMetaPartition(mp)
}
}
}
func (mp *MetaPartition) canBeOffline(nodeAddr string, replicaNum int) (err error) {

147
master/metadata_fsm_op.go
View File

@ -35,82 +35,84 @@ import (
transferred over the network. */
type clusterValue struct {
Name string
CreateTime int64
Threshold float32
LoadFactor float32
DisableAutoAllocate bool
ForbidMpDecommission bool
DataNodeDeleteLimitRate uint64
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeAutoRepairLimitRate uint64
MaxDpCntLimit uint64
MaxMpCntLimit uint64
FaultDomain bool
DiskQosEnable bool
QosLimitUpload uint64
DirChildrenNumLimit uint32
DecommissionLimit uint64
CheckDataReplicasEnable bool
FileStatsEnable bool
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt uint32
DataPartitionTimeoutSec int64
ForbidWriteOpOfProtoVer0 bool
LegacyDataMediaType uint32
Name string
CreateTime int64
Threshold float32
LoadFactor float32
DisableAutoAllocate bool
ForbidMpDecommission bool
DataNodeDeleteLimitRate uint64
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeAutoRepairLimitRate uint64
MaxDpCntLimit uint64
MaxMpCntLimit uint64
FaultDomain bool
DiskQosEnable bool
QosLimitUpload uint64
DirChildrenNumLimit uint32
DecommissionLimit uint64
CheckDataReplicasEnable bool
FileStatsEnable bool
ClusterUuid string
ClusterUuidEnable bool
MetaPartitionInodeIdStep uint64
MaxConcurrentLcNodes uint64
DpMaxRepairErrCnt uint64
DpRepairTimeOut uint64
DpBackupTimeOut uint64
EnableAutoDecommissionDisk bool
AutoDecommissionDiskInterval int64
DecommissionDiskLimit uint32
VolDeletionDelayTimeHour int64
MarkDiskBrokenThreshold float64
EnableAutoDpMetaRepair bool
AutoDpMetaRepairParallelCnt uint32
DataPartitionTimeoutSec int64
ForbidWriteOpOfProtoVer0 bool
LegacyDataMediaType uint32
RaftPartitionAlreadyUseDifferentPort bool
}
func newClusterValue(c *Cluster) (cv *clusterValue) {
cv = &clusterValue{
Name: c.Name,
CreateTime: c.CreateTime,
LoadFactor: c.cfg.ClusterLoadFactor,
Threshold: c.cfg.MetaNodeThreshold,
DataNodeDeleteLimitRate: c.cfg.DataNodeDeleteLimitRate,
MetaNodeDeleteBatchCount: c.cfg.MetaNodeDeleteBatchCount,
MetaNodeDeleteWorkerSleepMs: c.cfg.MetaNodeDeleteWorkerSleepMs,
DataNodeAutoRepairLimitRate: c.cfg.DataNodeAutoRepairLimitRate,
DisableAutoAllocate: c.DisableAutoAllocate,
ForbidMpDecommission: c.ForbidMpDecommission,
MaxDpCntLimit: c.cfg.MaxDpCntLimit,
MaxMpCntLimit: c.cfg.MaxMpCntLimit,
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
DpBackupTimeOut: c.cfg.DpBackupTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk.Load(),
AutoDecommissionDiskInterval: c.AutoDecommissionInterval.Load(),
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: c.AutoDpMetaRepairParallelCnt.Load(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
ForbidWriteOpOfProtoVer0: c.cfg.forbidWriteOpOfProtoVer0,
LegacyDataMediaType: c.legacyDataMediaType,
Name: c.Name,
CreateTime: c.CreateTime,
LoadFactor: c.cfg.ClusterLoadFactor,
Threshold: c.cfg.MetaNodeThreshold,
DataNodeDeleteLimitRate: c.cfg.DataNodeDeleteLimitRate,
MetaNodeDeleteBatchCount: c.cfg.MetaNodeDeleteBatchCount,
MetaNodeDeleteWorkerSleepMs: c.cfg.MetaNodeDeleteWorkerSleepMs,
DataNodeAutoRepairLimitRate: c.cfg.DataNodeAutoRepairLimitRate,
DisableAutoAllocate: c.DisableAutoAllocate,
ForbidMpDecommission: c.ForbidMpDecommission,
MaxDpCntLimit: c.cfg.MaxDpCntLimit,
MaxMpCntLimit: c.cfg.MaxMpCntLimit,
FaultDomain: c.FaultDomain,
DiskQosEnable: c.diskQosEnable,
QosLimitUpload: uint64(c.QosAcceptLimit.Limit()),
DirChildrenNumLimit: c.cfg.DirChildrenNumLimit,
DecommissionLimit: c.DecommissionLimit,
CheckDataReplicasEnable: c.checkDataReplicasEnable,
FileStatsEnable: c.fileStatsEnable,
ClusterUuid: c.clusterUuid,
ClusterUuidEnable: c.clusterUuidEnable,
MetaPartitionInodeIdStep: c.cfg.MetaPartitionInodeIdStep,
MaxConcurrentLcNodes: c.cfg.MaxConcurrentLcNodes,
DpMaxRepairErrCnt: c.cfg.DpMaxRepairErrCnt,
DpRepairTimeOut: c.cfg.DpRepairTimeOut,
DpBackupTimeOut: c.cfg.DpBackupTimeOut,
EnableAutoDecommissionDisk: c.EnableAutoDecommissionDisk.Load(),
AutoDecommissionDiskInterval: c.AutoDecommissionInterval.Load(),
DecommissionDiskLimit: c.GetDecommissionDiskLimit(),
VolDeletionDelayTimeHour: c.cfg.volDelayDeleteTimeHour,
MarkDiskBrokenThreshold: c.getMarkDiskBrokenThreshold(),
EnableAutoDpMetaRepair: c.getEnableAutoDpMetaRepair(),
AutoDpMetaRepairParallelCnt: c.AutoDpMetaRepairParallelCnt.Load(),
DataPartitionTimeoutSec: c.getDataPartitionTimeoutSec(),
ForbidWriteOpOfProtoVer0: c.cfg.forbidWriteOpOfProtoVer0,
LegacyDataMediaType: c.legacyDataMediaType,
RaftPartitionAlreadyUseDifferentPort: c.cfg.raftPartitionAlreadyUseDifferentPort.Load(),
}
return cv
}
@ -1306,6 +1308,7 @@ func (c *Cluster) loadClusterValue() (err error) {
c.updateEnableAutoDpMetaRepair(cv.EnableAutoDpMetaRepair)
c.updateAutoDpMetaRepairParallelCnt(cv.AutoDpMetaRepairParallelCnt)
c.updateDataPartitionTimeoutSec(cv.DataPartitionTimeoutSec)
c.cfg.raftPartitionAlreadyUseDifferentPort.Store(cv.RaftPartitionAlreadyUseDifferentPort)
c.cfg.forbidWriteOpOfProtoVer0 = cv.ForbidWriteOpOfProtoVer0
c.legacyDataMediaType = cv.LegacyDataMediaType
log.LogInfof("action[loadClusterValue] ForbidWriteOpOfProtoVer0(%v), mediaType %d",

2
master/node_selector.go
View File

@ -511,7 +511,7 @@ func (s *StrawNodeSelector) Select(ns *nodeSet, excludeHosts []string, replicaNu
})
if len(nodes) < replicaNum {
err = fmt.Errorf("action[%vNodeSelector::Select] no enough writable hosts,replicaNum:%v MatchNodeCount:%v ",
err = fmt.Errorf("action[%vNodeSelector-Select] no enough writable hosts,replicaNum:%v MatchNodeCount:%v ",
s.GetName(), replicaNum, len(nodes))
return
}

26
master/server.go
View File

@ -138,6 +138,13 @@ func (m *Server) Start(cfg *config.Config) (err error) {
m.config = newClusterConfig()
gConfig = m.config
m.leaderInfo = &LeaderInfo{}
m.storeDir = cfg.GetString(StoreDir)
if m.storeDir == "" {
return fmt.Errorf("store dir is empty")
}
if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil {
return
}
if err = m.checkConfig(cfg); err != nil {
log.LogError(errors.Stack(err))
@ -146,10 +153,6 @@ func (m *Server) Start(cfg *config.Config) (err error) {
m.reverseProxy = m.newReverseProxy()
m.cliMgr = newClientMgr()
if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil {
return
}
if err = m.createRaftServer(cfg); err != nil {
log.LogError(errors.Stack(err))
return
@ -232,7 +235,6 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) {
m.port = cfg.GetString(proto.ListenPort)
m.logDir = cfg.GetString(LogDir)
m.walDir = cfg.GetString(WalDir)
m.storeDir = cfg.GetString(StoreDir)
m.bStoreAddr = cfg.GetString(BStoreAddrKey)
if m.bStoreAddr == "" {
m.bStoreAddr = cfg.GetString(EbsAddrKey)
@ -242,9 +244,9 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) {
m.servicePath = cfg.GetString(EbsServicePathKey)
}
peerAddrs := cfg.GetString(cfgPeers)
if m.port == "" || m.walDir == "" || m.storeDir == "" || m.clusterName == "" || peerAddrs == "" {
return fmt.Errorf("%v,err:%v,%v,%v,%v,%v,%v", proto.ErrInvalidCfg, "one of (listen,walDir,storeDir,clusterName) is null",
m.port, m.walDir, m.storeDir, m.clusterName, peerAddrs)
if m.port == "" || m.walDir == "" || m.clusterName == "" || peerAddrs == "" {
return fmt.Errorf("%v,err:%v,%v,%v,%v,%v", proto.ErrInvalidCfg, "one of (listen,walDir,clusterName) is null",
m.port, m.walDir, m.clusterName, peerAddrs)
}
if m.id, err = strconv.ParseUint(cfg.GetString(ID), 10, 64); err != nil {
@ -397,12 +399,10 @@ 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
if err = m.config.checkRaftPartitionCanUseDifferentPort(m, cfg.GetBoolWithDefault(cfgRaftPartitionCanUseDifferentPort, false)); err != nil {
return err
}
m.config.AllowMultipleReplicasOnSameMachine = cfg.GetBoolWithDefault(cfgAllowMultipleReplicasOnSameMachine, true)
m.config.cfgDataMediaType = uint32(cfg.GetInt64(cfgLegacyDataMediaType))
if m.config.cfgDataMediaType != 0 && !proto.IsValidMediaType(m.config.cfgDataMediaType) {

2
metanode/metanode.go
View File

@ -483,7 +483,7 @@ func (m *MetaNode) register() (err error) {
m.clusterEnableSnapshot = gClusterInfo.ClusterEnableSnapshot
clusterEnableSnapshot = m.clusterEnableSnapshot
m.clusterId = gClusterInfo.Cluster
m.raftPartitionCanUsingDifferentPort = clusterInfo.RaftPartitionCanUsingDifferentPort
m.raftPartitionCanUsingDifferentPort = gClusterInfo.RaftPartitionCanUsingDifferentPort
nodeAddress = m.localAddr + ":" + m.listen
var settingsFromMaster *proto.UpgradeCompatibleSettings

2
metanode/partition.go
View File

@ -1071,7 +1071,7 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP
enableAuditLog: true,
}
if mp.manager.metaNode.raftPartitionCanUsingDifferentPort {
if mp.manager != nil && mp.manager.metaNode.raftPartitionCanUsingDifferentPort {
// during upgrade process, create partition request may lack raft ports info
defaultHeartbeatPort, defaultReplicaPort, err := mp.getRaftPort()
if err == nil {