diff --git a/docs-zh/source/ops/configs/master.md b/docs-zh/source/ops/configs/master.md index d727b16ce..43ea659dc 100755 --- a/docs-zh/source/ops/configs/master.md +++ b/docs-zh/source/ops/configs/master.md @@ -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 | ## 配置示例 diff --git a/docs/source/ops/configs/master.md b/docs/source/ops/configs/master.md index 0fdd6bfb4..e326d2aac 100755 --- a/docs/source/ops/configs/master.md +++ b/docs/source/ops/configs/master.md @@ -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 diff --git a/master/cluster.go b/master/cluster.go index 55b84b435..6f70e1527 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -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) diff --git a/master/config.go b/master/config.go index 6c4d7cab3..e6ee6b8c7 100644 --- a/master/config.go +++ b/master/config.go @@ -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 -} diff --git a/master/data_partition.go b/master/data_partition.go index 4651db0eb..233676742 100644 --- a/master/data_partition.go +++ b/master/data_partition.go @@ -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 } diff --git a/master/meta_partition.go b/master/meta_partition.go index d5704af49..69e7f4f8b 100644 --- a/master/meta_partition.go +++ b/master/meta_partition.go @@ -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) { diff --git a/master/metadata_fsm_op.go b/master/metadata_fsm_op.go index 64a54d1cc..e2ea0ec1a 100644 --- a/master/metadata_fsm_op.go +++ b/master/metadata_fsm_op.go @@ -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", diff --git a/master/node_selector.go b/master/node_selector.go index 21e1e8209..6eaad0a9b 100644 --- a/master/node_selector.go +++ b/master/node_selector.go @@ -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 } diff --git a/master/server.go b/master/server.go index cc7004405..598cf90b7 100644 --- a/master/server.go +++ b/master/server.go @@ -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) { diff --git a/metanode/metanode.go b/metanode/metanode.go index 32671a5c8..57b88c6db 100644 --- a/metanode/metanode.go +++ b/metanode/metanode.go @@ -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 diff --git a/metanode/partition.go b/metanode/partition.go index 425c882c9..897be9d44 100644 --- a/metanode/partition.go +++ b/metanode/partition.go @@ -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 {