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(datanode): support multi datanode process using different port on single node

Browse Source 
Signed-off-by: morphes1995 <morphes1995@gmail.com>
This commit is contained in:
morphes1995 2024-09-25 14:03:54 +08:00 committed by leonrayang
parent 2beb480989
commit b23df49eca
18 changed files with 155 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
blobstore/proxy/service.go
View File

@ -186,7 +186,7 @@ func NewHandler(service *Service) *rpc.Router {
// POST /volume/alloc
// request body: json
// response body: json
// response body: json/volume/alloc
router.Handle(http.MethodPost, "/volume/alloc", service.Alloc, rpc.OptArgsBody())
// GET /volume/list?code_mode={code_mode}

2
cli/cmd/fmt.go
View File

@ -940,6 +940,8 @@ func formatDataNodeDetail(dn *proto.DataNodeInfo, rowTable bool) string {
sb := strings.Builder{}
sb.WriteString(fmt.Sprintf(" ID : %v\n", dn.ID))
sb.WriteString(fmt.Sprintf(" Address : %v\n", formatAddr(dn.Addr, dn.DomainAddr)))
sb.WriteString(fmt.Sprintf(" RaftHeartbeatPort : %v\n", dn.RaftHeartbeatPort))
sb.WriteString(fmt.Sprintf(" RaftReplicaPort : %v\n", dn.RaftReplicaPort))
sb.WriteString(fmt.Sprintf(" Allocated ratio : %v\n", dn.UsageRatio))
sb.WriteString(fmt.Sprintf(" Allocated : %v\n", formatSize(dn.Used)))
sb.WriteString(fmt.Sprintf(" Available : %v\n", formatSize(dn.AvailableSpace)))

21
datanode/partition.go
View File

@ -273,6 +273,18 @@ func LoadDataPartition(partitionDir string, disk *Disk) (dp *DataPartition, err
if err = json.Unmarshal(metaFileData, meta); err != nil {
return
}
// compat old persisted data partitions, add raft port info
lackRaftPort := false
for i, peer := range meta.Peers {
if len(peer.ReplicaPort) == 0 || len(peer.HeartbeatPort) == 0 {
peer.ReplicaPort = disk.dataNode.raftReplica
peer.HeartbeatPort = disk.dataNode.raftHeartbeat
meta.Peers[i] = peer
lackRaftPort = true
}
}
if err = meta.Validate(); err != nil {
return
}
@ -349,6 +361,15 @@ func LoadDataPartition(partitionDir string, disk *Disk) (dp *DataPartition, err
disk.doDiskError()
}
}
if lackRaftPort {
// persist dp meta with raft port
if err = dp.PersistMetadata(); err != nil {
log.LogErrorf("persist dp(%v) meta with raft port error(%v)", dp.partitionID, err)
return
}
}
return
}

19
datanode/partition_raft.go
View File

@ -109,11 +109,13 @@ func (dp *DataPartition) StartRaft(isLoad bool) (err error) {
}
}()
if heartbeatPort, replicaPort, err = dp.raftPort(); err != nil {
return
}
//if heartbeatPort, replicaPort, 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)
rp := raftstore.PeerAddress{
Peer: raftproto.Peer{
ID: peer.ID,
@ -410,9 +412,16 @@ func (dp *DataPartition) addRaftNode(req *proto.AddDataPartitionRaftMemberReques
heartbeatPort int
replicaPort int
)
if heartbeatPort, replicaPort, err = dp.raftPort(); err != nil {
heartbeatPort, err = strconv.Atoi(req.AddPeer.HeartbeatPort)
if err != nil {
return
}
replicaPort, err = strconv.Atoi(req.AddPeer.ReplicaPort)
if err != nil {
return
}
log.LogInfof("action[addRaftNode] add raft node peer [%v]", req.AddPeer)
found := false
for _, peer := range dp.config.Peers {
@ -570,8 +579,6 @@ func (s *DataNode) parseRaftConfig(cfg *config.Config) (err error) {
func (s *DataNode) startRaftServer(cfg *config.Config) (err error) {
log.LogInfo("Start: startRaftServer")
s.parseRaftConfig(cfg)
if s.clusterUuidEnable {
if err = config.CheckOrStoreClusterUuid(s.raftDir, s.clusterUuid, false); err != nil {
log.LogErrorf("CheckOrStoreClusterUuid failed: %v", err)

5
datanode/server.go
View File

@ -254,6 +254,9 @@ func doStart(server common.Server, cfg *config.Config) (err error) {
if err = s.parseConfig(cfg); err != nil {
return
}
if err = s.parseRaftConfig(cfg); err != nil {
return
}
s.registerMetrics()
@ -711,7 +714,7 @@ func (s *DataNode) register(cfg *config.Config) (err error) {
// register this data node on the master
var nodeID uint64
if nodeID, err = MasterClient.NodeAPI().AddDataNodeWithAuthNode(fmt.Sprintf("%s:%v", LocalIP, s.port),
if nodeID, err = MasterClient.NodeAPI().AddDataNodeWithAuthNode(fmt.Sprintf("%s:%v", LocalIP, s.port), s.raftHeartbeat, s.raftReplica,
s.zoneName, s.serviceIDKey, s.mediaType); err != nil {
if strings.Contains(err.Error(), proto.ErrDataNodeAdd.Error()) {
failMsg := fmt.Sprintf("[register] register to master[%v] failed: %v",

2
datanode/space_manager.go
View File

@ -312,7 +312,7 @@ func (manager *SpaceManager) LoadDisk(path string, reservedSpace, diskRdonlySpac
log.LogDebugf("action[LoadDisk] load disk from path(%v).", path)
visitor = func(dp *DataPartition) {
// do noting here, dp is attached to space manager in RestorePartition
// do nothing here, dp is attached to space manager in RestorePartition
}
if _, err = manager.GetDisk(path); err != nil {

12
master/api_args_parse.go
View File

@ -19,6 +19,7 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/cubefs/cubefs/raftstore"
"io"
"math"
"net/http"
@ -165,7 +166,7 @@ func parseRequestForUpdateMetaNode(r *http.Request) (nodeAddr string, id uint64,
return
}
func parseRequestForAddNode(r *http.Request) (nodeAddr, zoneName string, mediaType uint32, err error) {
func parseRequestForAddNode(r *http.Request) (nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName string, mediaType uint32, err error) {
if err = r.ParseForm(); err != nil {
return
}
@ -175,6 +176,15 @@ func parseRequestForAddNode(r *http.Request) (nodeAddr, zoneName string, mediaTy
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)
}
if mediaType, err = extractMediaType(r); err != nil {
return
}

24
master/api_service.go
View File

@ -3180,19 +3180,21 @@ func checkIpPort(addr string) bool {
func (m *Server) addDataNode(w http.ResponseWriter, r *http.Request) {
var (
nodeAddr string
zoneName string
mediaType uint32
id uint64
err error
nodesetId uint64
nodeAddr string
zoneName string
raftHeartbeatPort string
raftReplicaPort string
mediaType uint32
id uint64
err error
nodesetId uint64
)
metric := exporter.NewTPCnt(apiToMetricsName(proto.AddDataNode))
defer func() {
doStatAndMetric(proto.AddDataNode, metric, err, nil)
}()
if nodeAddr, zoneName, mediaType, err = parseRequestForAddNode(r); err != nil {
if nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName, mediaType, err = parseRequestForAddNode(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
@ -3209,7 +3211,7 @@ func (m *Server) addDataNode(w http.ResponseWriter, r *http.Request) {
return
}
}
if id, err = m.cluster.addDataNode(nodeAddr, zoneName, nodesetId, mediaType); err != nil {
if id, err = m.cluster.addDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName, nodesetId, mediaType); err != nil {
log.LogErrorf("addDataNode: add failed, addr %s, zone %s, set %d, type %d, err %s",
nodeAddr, zoneName, nodesetId, mediaType, err.Error())
err = errors.NewErrorf("add datanode failed, err %s, hint %s", err.Error(), proto.ErrDataNodeAdd.Error())
@ -3250,6 +3252,8 @@ func (m *Server) getDataNode(w http.ResponseWriter, r *http.Request) {
ID: dataNode.ID,
ZoneName: dataNode.ZoneName,
Addr: dataNode.Addr,
RaftHeartbeatPort: dataNode.HeartbeatPort,
RaftReplicaPort: dataNode.ReplicaPort,
DomainAddr: dataNode.DomainAddr,
ReportTime: dataNode.ReportTime,
IsActive: dataNode.isActive,
@ -4880,7 +4884,7 @@ func (m *Server) addMetaNode(w http.ResponseWriter, r *http.Request) {
doStatAndMetric(proto.AddMetaNode, metric, err, nil)
}()
if nodeAddr, zoneName, _, err = parseRequestForAddNode(r); err != nil {
if nodeAddr, _, _, zoneName, _, err = parseRequestForAddNode(r); err != nil {
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
return
}
@ -7884,7 +7888,7 @@ func (m *Server) recoverBackupDataReplica(w http.ResponseWriter, r *http.Request
}
defer dp.setRestoreReplicaStop()
// restore raft member first
addPeer := proto.Peer{ID: dataNode.ID, Addr: addr}
addPeer := proto.Peer{ID: dataNode.ID, Addr: addr, HeartbeatPort: dataNode.HeartbeatPort, ReplicaPort: dataNode.ReplicaPort}
log.LogInfof("action[recoverBackupDataReplica] dp %v dst addr %v try add raft member, node id %v", dp.PartitionID, addr, dataNode.ID)
if err = m.cluster.addDataPartitionRaftMember(dp, addPeer); err != nil {

14
master/cluster.go
View File

@ -1238,7 +1238,7 @@ func (c *Cluster) checkSetZoneMediaTypePersist(zone *Zone, mediaType uint32) (ch
return true, nil
}
func (c *Cluster) addDataNode(nodeAddr, zoneName string, nodesetId uint64, mediaType uint32) (id uint64, err error) {
func (c *Cluster) addDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName string, nodesetId uint64, mediaType uint32) (id uint64, err error) {
c.dnMutex.Lock()
defer c.dnMutex.Unlock()
var dataNode *DataNode
@ -1276,11 +1276,17 @@ func (c *Cluster) addDataNode(nodeAddr, zoneName string, nodesetId uint64, media
if mediaType != dataNode.MediaType {
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
dataNode.HeartbeatPort = raftHeartbeatPort
dataNode.ReplicaPort = raftReplicaPort
c.syncUpdateDataNode(dataNode)
return dataNode.ID, nil
}
needPersistZone := false
dataNode = newDataNode(nodeAddr, zoneName, c.Name, mediaType)
dataNode = newDataNode(nodeAddr, raftHeartbeatPort, raftReplicaPort, zoneName, c.Name, mediaType)
dataNode.DpCntLimit = newLimitCounter(&c.cfg.MaxDpCntLimit, defaultMaxDpCntLimit)
if zone, _ = c.t.getZone(zoneName); zone == nil {
log.LogInfof("[addDataNode] create zone(%v) by datanode(%v), mediaType(%v)",
@ -2914,7 +2920,7 @@ func (c *Cluster) addDataReplica(dp *DataPartition, addr string, ignoreDecommiss
return
}
addPeer := proto.Peer{ID: targetDataNode.ID, Addr: addr}
addPeer := proto.Peer{ID: targetDataNode.ID, Addr: addr, HeartbeatPort: targetDataNode.HeartbeatPort, ReplicaPort: targetDataNode.ReplicaPort}
if !proto.IsNormalDp(dp.PartitionType) {
return fmt.Errorf("action[addDataReplica] [%d] is not normal dp, not support add or delete replica", dp.PartitionID)
@ -3141,7 +3147,7 @@ func (c *Cluster) removeDataReplica(dp *DataPartition, addr string, validate boo
if !proto.IsNormalDp(dp.PartitionType) {
return fmt.Errorf("[%d] is not normal dp, not support add or delete replica", dp.PartitionID)
}
removePeer := proto.Peer{ID: dataNode.ID, Addr: addr}
removePeer := proto.Peer{ID: dataNode.ID, Addr: addr, HeartbeatPort: dataNode.HeartbeatPort, ReplicaPort: dataNode.ReplicaPort}
if err = c.removeDataPartitionRaftMember(dp, removePeer, raftForceDel); err != nil {
return
}

18
master/data_node.go
View File

@ -39,6 +39,8 @@ type DataNode struct {
ID uint64
ZoneName string `json:"Zone"`
Addr string
HeartbeatPort string `json:"HeartbeatPort"`
ReplicaPort string `json:"ReplicaPort"`
DomainAddr string
ReportTime time.Time
StartTime int64
@ -83,7 +85,7 @@ type DataNode struct {
DpOpLogs []proto.OpLog
}
func newDataNode(addr, zoneName, clusterID string, mediaType uint32) (dataNode *DataNode) {
func newDataNode(addr, raftHeartbeatPort, raftReplicaPort, zoneName, clusterID string, mediaType uint32) (dataNode *DataNode) {
if zoneName == "" {
zoneName = DefaultZoneName
}
@ -91,6 +93,8 @@ func newDataNode(addr, zoneName, clusterID string, mediaType uint32) (dataNode *
dataNode = new(DataNode)
dataNode.Total = 1
dataNode.Addr = addr
dataNode.HeartbeatPort = raftHeartbeatPort
dataNode.ReplicaPort = raftReplicaPort
dataNode.ZoneName = zoneName
dataNode.LastUpdateTime = time.Now().Add(-time.Minute)
dataNode.TaskManager = newAdminTaskManager(dataNode.Addr, clusterID)
@ -394,6 +398,18 @@ func (dataNode *DataNode) isWriteAbleWithSizeNoLock(size uint64) (ok bool) {
return
}
func (dataNode *DataNode) GetHeartbeatPort() string {
dataNode.RLock()
defer dataNode.RUnlock()
return dataNode.HeartbeatPort
}
func (dataNode *DataNode) GetReplicaPort() string {
dataNode.RLock()
defer dataNode.RUnlock()
return dataNode.ReplicaPort
}
func (dataNode *DataNode) GetUsed() uint64 {
dataNode.RLock()
defer dataNode.RUnlock()

2
master/data_partition.go
View File

@ -2368,7 +2368,7 @@ func (partition *DataPartition) removeHostByForce(c *Cluster, peerAddr string) {
partition.PartitionID, peerAddr, err)
return
}
removePeer := proto.Peer{ID: dataNode.ID, Addr: peerAddr}
removePeer := proto.Peer{ID: dataNode.ID, Addr: peerAddr, HeartbeatPort: dataNode.HeartbeatPort, ReplicaPort: dataNode.ReplicaPort}
if err = c.removeHostMember(partition, removePeer); err != nil {
log.LogWarnf("action[removeHostByForce]dp %v remove host %v failed:%v",
partition.PartitionID, peerAddr, err)

61
master/meta_node.go
View File

@ -27,30 +27,31 @@ import (
// MetaNode defines the structure of a meta node
type MetaNode struct {
ID uint64
Addr string
DomainAddr string
IsActive bool
Sender *AdminTaskManager `graphql:"-"`
ZoneName string `json:"Zone"`
MaxMemAvailWeight uint64 `json:"MaxMemAvailWeight"`
Total uint64 `json:"TotalWeight"`
Used uint64 `json:"UsedWeight"`
Ratio float64
SelectCount uint64
Threshold float32
ReportTime time.Time
metaPartitionInfos []*proto.MetaPartitionReport
MetaPartitionCount int
NodeSetID uint64
sync.RWMutex `graphql:"-"`
ToBeOffline bool
PersistenceMetaPartitions []uint64
RdOnly bool
MigrateLock sync.RWMutex
MpCntLimit LimitCounter `json:"-"` // max count of meta partition in a meta node
CpuUtil atomicutil.Float64 `json:"-"`
ID uint64
Addr string
DomainAddr string
IsActive bool
Sender *AdminTaskManager `graphql:"-"`
ZoneName string `json:"Zone"`
MaxMemAvailWeight uint64 `json:"MaxMemAvailWeight"`
Total uint64 `json:"TotalWeight"`
Used uint64 `json:"UsedWeight"`
Ratio float64
SelectCount uint64
Threshold float32
ReportTime time.Time
metaPartitionInfos []*proto.MetaPartitionReport
MetaPartitionCount int
NodeSetID uint64
sync.RWMutex `graphql:"-"`
ToBeOffline bool
PersistenceMetaPartitions []uint64
RdOnly bool
MigrateLock sync.RWMutex
MpCntLimit LimitCounter `json:"-"` // max count of meta partition in a meta node
CpuUtil atomicutil.Float64 `json:"-"`
HeartbeatPort string `json:"HeartbeatPort"`
ReplicaPort string `json:"ReplicaPort"`
ReceivedForbidWriteOpOfProtoVer0 bool
}
@ -100,6 +101,18 @@ func (metaNode *MetaNode) GetID() uint64 {
return metaNode.ID
}
func (metaNode *MetaNode) GetHeartbeatPort() string {
metaNode.RLock()
defer metaNode.RUnlock()
return metaNode.HeartbeatPort
}
func (metaNode *MetaNode) GetReplicaPort() string {
metaNode.RLock()
defer metaNode.RUnlock()
return metaNode.ReplicaPort
}
func (metaNode *MetaNode) GetAddr() string {
metaNode.RLock()
defer metaNode.RUnlock()

6
master/metadata_fsm_op.go
View File

@ -451,6 +451,8 @@ type dataNodeValue struct {
ID uint64
NodeSetID uint64
Addr string
HeartbeatPort string
ReplicaPort string
ZoneName string
RdOnly bool
DecommissionedDisks []string
@ -471,6 +473,8 @@ func newDataNodeValue(dataNode *DataNode) *dataNodeValue {
ID: dataNode.ID,
NodeSetID: dataNode.NodeSetID,
Addr: dataNode.Addr,
HeartbeatPort: dataNode.HeartbeatPort,
ReplicaPort: dataNode.ReplicaPort,
ZoneName: dataNode.ZoneName,
RdOnly: dataNode.RdOnly,
DecommissionedDisks: dataNode.getDecommissionedDisks(),
@ -1519,7 +1523,7 @@ func (c *Cluster) loadDataNodes() (err error) {
dnv.Addr, proto.MediaTypeString(dnv.MediaType))
}
dataNode := newDataNode(dnv.Addr, dnv.ZoneName, c.Name, dnv.MediaType)
dataNode := newDataNode(dnv.Addr, dnv.HeartbeatPort, dnv.ReplicaPort, dnv.ZoneName, c.Name, dnv.MediaType)
dataNode.DpCntLimit = newLimitCounter(&c.cfg.MaxDpCntLimit, defaultMaxDpCntLimit)
dataNode.ID = dnv.ID
dataNode.NodeSetID = dnv.NodeSetID

10
master/node_selector.go
View File

@ -65,6 +65,8 @@ type Node interface {
SelectNodeForWrite()
GetID() uint64
GetAddr() string
GetHeartbeatPort() string
GetReplicaPort() string
PartitionCntLimited() bool
IsActiveNode() bool
IsWriteAble() bool
@ -238,7 +240,7 @@ func (s *CarryWeightNodeSelector) Select(ns *nodeSet, excludeHosts []string, rep
node := weightedNodes[i].Ptr
s.selectNodeForWrite(node)
orderHosts = append(orderHosts, node.GetAddr())
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr()}
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr(), ReplicaPort: node.GetReplicaPort(), HeartbeatPort: node.GetHeartbeatPort()}
peers = append(peers, peer)
}
log.LogInfof("action[%vNodeSelector-Select] peers[%v]", s.GetName(), peers)
@ -313,7 +315,7 @@ func (s *AvailableSpaceFirstNodeSelector) Select(ns *nodeSet, excludeHosts []str
node := sortedNodes[selectedIndex]
node.SelectNodeForWrite()
orderHosts = append(orderHosts, node.GetAddr())
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr()}
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr(), ReplicaPort: node.GetReplicaPort(), HeartbeatPort: node.GetHeartbeatPort()}
peers = append(peers, peer)
}
}
@ -392,7 +394,7 @@ func (s *RoundRobinNodeSelector) Select(ns *nodeSet, excludeHosts []string, repl
node := sortedNodes[(selectedIndex+s.index)%len(sortedNodes)]
orderHosts = append(orderHosts, node.GetAddr())
node.SelectNodeForWrite()
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr()}
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr(), ReplicaPort: node.GetReplicaPort(), HeartbeatPort: node.GetHeartbeatPort()}
peers = append(peers, peer)
}
}
@ -476,7 +478,7 @@ func (s *StrawNodeSelector) Select(ns *nodeSet, excludeHosts []string, replicaNu
}
orderHosts = append(orderHosts, node.GetAddr())
node.SelectNodeForWrite()
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr()}
peer := proto.Peer{ID: node.GetID(), Addr: node.GetAddr(), ReplicaPort: node.GetReplicaPort(), HeartbeatPort: node.GetHeartbeatPort()}
peers = append(peers, peer)
}
// if we cannot get enough writable nodes, return error

4
master/topology_test.go
View File

@ -1,6 +1,8 @@
package master
import (
"github.com/cubefs/cubefs/raftstore"
"strconv"
"testing"
"time"
@ -11,7 +13,7 @@ import (
)
func createDataNodeForTopo(addr, zoneName string, ns *nodeSet) (dn *DataNode) {
dn = newDataNode(addr, zoneName, "test", proto.MediaType_HDD)
dn = newDataNode(addr, strconv.Itoa(raftstore.DefaultHeartbeatPort), strconv.Itoa(raftstore.DefaultReplicaPort), zoneName, "test", proto.MediaType_HDD)
dn.ZoneName = zoneName
dn.Total = 1024 * util.GB
dn.Used = 10 * util.GB

6
proto/meta_proto.go
View File

@ -34,8 +34,10 @@ type CreateNameSpaceResponse struct {
// Peer defines the peer of the node id and address.
type Peer struct {
ID uint64 `json:"id"`
Addr string `json:"addr"`
ID uint64 `json:"id"`
Addr string `json:"addr"`
HeartbeatPort string `json:"raftHeartbeat"`
ReplicaPort string `json:"raftReplica"`
}
// CreateMetaPartitionRequest defines the request to create a meta partition.

2
proto/model.go
View File

@ -58,6 +58,8 @@ type DataNodeInfo struct {
ID uint64
ZoneName string `json:"Zone"`
Addr string
RaftHeartbeatPort string
RaftReplicaPort string
DomainAddr string
ReportTime time.Time
IsActive bool

4
sdk/master/api_node.go
View File

@ -50,9 +50,11 @@ func (api *NodeAPI) AddDataNode(serverAddr, zoneName string, mediaType uint32) (
return
}
func (api *NodeAPI) AddDataNodeWithAuthNode(serverAddr, zoneName, clientIDKey string, mediaType uint32) (id uint64, err error) {
func (api *NodeAPI) AddDataNodeWithAuthNode(serverAddr, raftHeartbeatPort, raftReplicaPort, zoneName, clientIDKey string, mediaType uint32) (id uint64, err error) {
request := newRequest(get, proto.AddDataNode).Header(api.h)
request.addParam("addr", serverAddr)
request.addParam("heartbeatPort", raftHeartbeatPort)
request.addParam("replicaPort", raftReplicaPort)
request.addParam("zoneName", zoneName)
request.addParam("clientIDKey", clientIDKey)
request.addParam("mediaType", strconv.Itoa(int(mediaType)))