mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(flashnode): add flash node service in master
. #22200531 #22115371 #22200563 Signed-off-by: slasher <mcq.sejust@gmail.com>
This commit is contained in:
slasher
zhumingze1108
parent
b150a10a36
commit
c4aaa43205
@ -894,12 +894,14 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
|
||||
ForbidWriteOpOfProtoVer0: m.cluster.cfg.forbidWriteOpOfProtoVer0,
|
||||
LegacyDataMediaType: m.cluster.legacyDataMediaType,
|
||||
RaftPartitionCanUsingDifferentPortEnabled: m.cluster.RaftPartitionCanUsingDifferentPortEnabled(),
|
||||
FlashNodes: make([]proto.NodeView, 0),
|
||||
}
|
||||
|
||||
vols := m.cluster.allVolNames()
|
||||
cv.MasterNodes = m.cluster.allMasterNodes()
|
||||
cv.MetaNodes = m.cluster.allMetaNodes()
|
||||
cv.DataNodes = m.cluster.allDataNodes()
|
||||
cv.FlashNodes = m.cluster.allFlashNodes()
|
||||
cv.DataNodeStatInfo = m.cluster.dataNodeStatInfo
|
||||
cv.MetaNodeStatInfo = m.cluster.metaNodeStatInfo
|
||||
for _, name := range vols {
|
||||
|
||||
@ -139,6 +139,9 @@ type Cluster struct {
|
||||
|
||||
ac *authSDK.AuthClient
|
||||
masterClient *masterSDK.MasterClient
|
||||
|
||||
flashNodeTopo *flashNodeTopology
|
||||
flashGroupRespCache atomic.Value // []byte
|
||||
}
|
||||
|
||||
type cTask struct {
|
||||
@ -430,6 +433,7 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
|
||||
c.EnableAutoDpMetaRepair.Store(defaultEnableDpMetaRepair)
|
||||
c.AutoDecommissionInterval.Store(int64(defaultAutoDecommissionDiskInterval))
|
||||
c.server = server
|
||||
c.flashNodeTopo = newFlashNodeTopology()
|
||||
return
|
||||
}
|
||||
|
||||
@ -456,6 +460,7 @@ func (c *Cluster) scheduleTask() {
|
||||
c.scheduleToBadDisk()
|
||||
c.scheduleToCheckVolUid()
|
||||
c.scheduleToCheckDataReplicaMeta()
|
||||
c.scheduleToUpdateFlashGroupRespCache()
|
||||
}
|
||||
|
||||
func (c *Cluster) masterAddr() (addr string) {
|
||||
@ -4102,6 +4107,21 @@ func (c *Cluster) allMetaNodes() (metaNodes []proto.NodeView) {
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) allFlashNodes() (flashNodes []proto.NodeView) {
|
||||
flashNodes = make([]proto.NodeView, 0)
|
||||
c.flashNodeTopo.flashNodeMap.Range(func(addr, node interface{}) bool {
|
||||
flashNode := node.(*FlashNode)
|
||||
flashNodes = append(flashNodes, proto.NodeView{
|
||||
ID: flashNode.ID,
|
||||
Addr: flashNode.Addr,
|
||||
Status: flashNode.IsActive,
|
||||
IsWritable: flashNode.isWriteAble(),
|
||||
})
|
||||
return true
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// get metaNode with specified condition
|
||||
func (c *Cluster) getSpecifiedMetaNodes(zones map[string]struct{}, nodeSetIds map[uint64]struct{}) (metaNodes []*MetaNode) {
|
||||
log.LogInfof("cluster metaNode length:%v", c.allMetaNodes())
|
||||
|
||||
@ -363,7 +363,8 @@ func TestBalanceMetaPartition(t *testing.T) {
|
||||
|
||||
func TestMasterClientLeaderChange(t *testing.T) {
|
||||
cluster := &Cluster{
|
||||
masterClient: masterSDK.NewMasterClient(nil, false),
|
||||
masterClient: masterSDK.NewMasterClient(nil, false),
|
||||
flashNodeTopo: newFlashNodeTopology(),
|
||||
}
|
||||
cluster.t = newTopology()
|
||||
cluster.BadDataPartitionIds = new(sync.Map)
|
||||
|
||||
@ -157,6 +157,10 @@ const (
|
||||
quotaClass = "quotaClass"
|
||||
quotaOfClass = "quotaOfStorageClass"
|
||||
dataMediaTypeKey = "dataMediaType"
|
||||
|
||||
stateKey = "state"
|
||||
versionKey = "version"
|
||||
fgSlotsKey = "slots"
|
||||
)
|
||||
|
||||
const (
|
||||
@ -240,6 +244,9 @@ const (
|
||||
unavailableZone = 1
|
||||
metaNodesUnAvailable = 2
|
||||
dataNodesUnAvailable = 3
|
||||
|
||||
unusedFlashNodeFlashGroupID = 0
|
||||
defaultFlashGroupSlotsCount = 128
|
||||
)
|
||||
|
||||
const (
|
||||
@ -310,6 +317,13 @@ const (
|
||||
opSyncAddLcResult uint32 = 0x3a
|
||||
opSyncDeleteLcResult uint32 = 0x3b
|
||||
|
||||
opSyncAddFlashNode uint32 = 0x3A
|
||||
opSyncDeleteFlashNode uint32 = 0x3B
|
||||
opSyncUpdateFlashNode uint32 = 0x3C
|
||||
opSyncAddFlashGroup uint32 = 0x3D
|
||||
opSyncDeleteFlashGroup uint32 = 0x3E
|
||||
opSyncUpdateFlashGroup uint32 = 0x3F
|
||||
|
||||
opSyncAllocQuotaID uint32 = 0x40
|
||||
opSyncSetQuota uint32 = 0x41
|
||||
opSyncDeleteQuota uint32 = 0x42
|
||||
@ -371,6 +385,9 @@ const (
|
||||
lcTaskPrefix = keySeparator + lcTaskAcronym + keySeparator
|
||||
lcResultPrefix = keySeparator + lcResultAcronym + keySeparator
|
||||
S3QoSPrefix = keySeparator + S3QoS + keySeparator
|
||||
|
||||
flashNodePrefix = keySeparator + "fn" + keySeparator
|
||||
flashGroupPrefix = keySeparator + "fg" + keySeparator
|
||||
)
|
||||
|
||||
// selector enum
|
||||
|
||||
611
master/flash_group.go
Normal file
611
master/flash_group.go
Normal file
@ -0,0 +1,611 @@
|
||||
// Copyright 2018 The CubeFS Authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied. See the License for the specific language governing
|
||||
// permissions and limitations under the License.
|
||||
|
||||
package master
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
type flashGroupValue struct {
|
||||
ID uint64
|
||||
Slots []uint32 // FlashGroup's position in hasher ring, set by cli. value is range of crc32.
|
||||
Status proto.FlashGroupStatus
|
||||
}
|
||||
|
||||
type FlashGroup struct {
|
||||
flashGroupValue
|
||||
lock sync.RWMutex
|
||||
flashNodes map[string]*FlashNode // key: FlashNodeAddr
|
||||
}
|
||||
|
||||
func newFlashGroup(id uint64, slots []uint32, status proto.FlashGroupStatus) *FlashGroup {
|
||||
fg := new(FlashGroup)
|
||||
fg.ID = id
|
||||
fg.Slots = slots
|
||||
fg.Status = status
|
||||
fg.flashNodes = make(map[string]*FlashNode)
|
||||
return fg
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) putFlashNode(fn *FlashNode) {
|
||||
fg.lock.Lock()
|
||||
fg.flashNodes[fn.Addr] = fn
|
||||
fg.lock.Unlock()
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) removeFlashNode(addr string) {
|
||||
fg.lock.Lock()
|
||||
delete(fg.flashNodes, addr)
|
||||
fg.lock.Unlock()
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) getTargetZoneFlashNodeHosts(targetZone string) (hosts []string) {
|
||||
fg.lock.RLock()
|
||||
for _, flashNode := range fg.flashNodes {
|
||||
if flashNode.ZoneName == targetZone {
|
||||
hosts = append(hosts, flashNode.Addr)
|
||||
}
|
||||
}
|
||||
fg.lock.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) getFlashNodeHosts(checkStatus bool) (hosts []string) {
|
||||
hosts = make([]string, 0, len(fg.flashNodes))
|
||||
fg.lock.RLock()
|
||||
for host, flashNode := range fg.flashNodes {
|
||||
if checkStatus && !flashNode.isActiveAndEnable() {
|
||||
continue
|
||||
}
|
||||
hosts = append(hosts, host)
|
||||
}
|
||||
fg.lock.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) getFlashNodesCount() (count int) {
|
||||
fg.lock.RLock()
|
||||
count = len(fg.flashNodes)
|
||||
fg.lock.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) syncAddFlashGroup(flashGroup *FlashGroup) (err error) {
|
||||
return c.syncPutFlashGroupInfo(opSyncAddFlashGroup, flashGroup)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncDeleteFlashGroup(flashGroup *FlashGroup) (err error) {
|
||||
return c.syncPutFlashGroupInfo(opSyncDeleteFlashGroup, flashGroup)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncUpdateFlashGroup(flashGroup *FlashGroup) (err error) {
|
||||
return c.syncPutFlashGroupInfo(opSyncUpdateFlashGroup, flashGroup)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncPutFlashGroupInfo(opType uint32, flashGroup *FlashGroup) (err error) {
|
||||
metadata := new(RaftCmd)
|
||||
metadata.Op = opType
|
||||
metadata.K = flashGroupPrefix + strconv.FormatUint(flashGroup.ID, 10)
|
||||
metadata.V, err = json.Marshal(flashGroup.flashGroupValue)
|
||||
if err != nil {
|
||||
return errors.New(err.Error())
|
||||
}
|
||||
return c.submit(metadata)
|
||||
}
|
||||
|
||||
func (fg *FlashGroup) GetAdminView() (view proto.FlashGroupAdminView) {
|
||||
view = proto.FlashGroupAdminView{
|
||||
ID: fg.ID,
|
||||
Slots: fg.Slots,
|
||||
Status: fg.Status,
|
||||
}
|
||||
view.ZoneFlashNodes = make(map[string][]*proto.FlashNodeViewInfo)
|
||||
fg.lock.RLock()
|
||||
view.FlashNodeCount = len(fg.flashNodes)
|
||||
for _, flashNode := range fg.flashNodes {
|
||||
view.ZoneFlashNodes[flashNode.ZoneName] = append(view.ZoneFlashNodes[flashNode.ZoneName], flashNode.getFlashNodeViewInfo())
|
||||
}
|
||||
fg.lock.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) createFlashGroup(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupCreate))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupCreate, metric, err, nil)
|
||||
}()
|
||||
setSlots := getSetSlots(r)
|
||||
flashGroup, err := m.cluster.createFlashGroup(setSlots)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView()))
|
||||
}
|
||||
|
||||
func (c *Cluster) createFlashGroup(setSlots []uint32) (fg *FlashGroup, err error) {
|
||||
defer func() {
|
||||
if err != nil {
|
||||
log.LogErrorf("action[addFlashGroup],clusterID[%v] err:%v ", c.Name, err.Error())
|
||||
}
|
||||
}()
|
||||
id, err := c.idAlloc.allocateCommonID()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if fg, err = c.flashNodeTopo.createFlashGroup(id, c, setSlots); err != nil {
|
||||
return
|
||||
}
|
||||
log.LogInfof("action[addFlashGroup],clusterID[%v] id:%v Slots:%v success", c.Name, fg.ID, fg.Slots)
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) removeFlashGroup(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
flashGroupID uint64
|
||||
flashGroup *FlashGroup
|
||||
needUpdateFGCache bool
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupRemove))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupRemove, metric, err, nil)
|
||||
}()
|
||||
if flashGroupID, err = extractFlashGroupID(r); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if flashGroup, err = m.cluster.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if flashGroup.Status == proto.FlashGroupStatus_Active && flashGroup.getFlashNodesCount() != 0 {
|
||||
needUpdateFGCache = true
|
||||
}
|
||||
if err = m.cluster.removeFlashGroup(flashGroup); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if needUpdateFGCache {
|
||||
m.cluster.updateFlashGroupResponseCache()
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("remove flashGroup:%v successfully,Slots:%v nodeCount:%v",
|
||||
flashGroup.ID, flashGroup.Slots, flashGroup.getFlashNodesCount())))
|
||||
}
|
||||
|
||||
func (c *Cluster) removeFlashGroup(flashGroup *FlashGroup) (err error) {
|
||||
// remove flash nodes then del the flash group
|
||||
flashNodeHosts := flashGroup.getFlashNodeHosts(false)
|
||||
successHost := make([]string, 0)
|
||||
for _, flashNodeHost := range flashNodeHosts {
|
||||
if err = c.removeFlashNodeFromFlashGroup(flashNodeHost, flashGroup); err != nil {
|
||||
err = fmt.Errorf("successHost:%v, flashNodeHosts:%v err:%v", successHost, flashNodeHosts, err)
|
||||
return
|
||||
}
|
||||
successHost = append(successHost, flashNodeHost)
|
||||
}
|
||||
log.LogInfo(fmt.Sprintf("action[removeFlashGroup] flashGroup:%v successHost:%v", flashGroup.ID, successHost))
|
||||
err = c.flashNodeTopo.removeFlashGroup(flashGroup, c)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) setFlashGroup(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
flashGroupID uint64
|
||||
fgStatus proto.FlashGroupStatus
|
||||
flashGroup *FlashGroup
|
||||
err error
|
||||
needUpdateFGCache bool
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupSet))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupSet, metric, err, nil)
|
||||
}()
|
||||
if flashGroupID, fgStatus, err = parseRequestForSetFlashGroup(r); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if flashGroup, err = m.cluster.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
oldStatus := flashGroup.Status
|
||||
flashGroup.Status = fgStatus
|
||||
if oldStatus == proto.FlashGroupStatus_Active && fgStatus == proto.FlashGroupStatus_Inactive {
|
||||
needUpdateFGCache = true
|
||||
}
|
||||
if err = m.cluster.syncUpdateFlashGroup(flashGroup); err != nil {
|
||||
flashGroup.Status = oldStatus
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if needUpdateFGCache {
|
||||
m.cluster.updateFlashGroupResponseCache()
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView()))
|
||||
}
|
||||
|
||||
func (m *Server) getFlashGroup(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
flashGroupID uint64
|
||||
flashGroup *FlashGroup
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupGet))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupGet, metric, err, nil)
|
||||
}()
|
||||
if flashGroupID, err = extractFlashGroupID(r); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if flashGroup, err = m.cluster.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView()))
|
||||
}
|
||||
|
||||
func (m *Server) flashGroupAddFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupNodeAdd))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupNodeAdd, metric, err, nil)
|
||||
}()
|
||||
|
||||
flashGroupID, addr, zoneName, count, err := parseRequestForManageFlashNodeOfFlashGroup(r)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
var flashGroup *FlashGroup
|
||||
if flashGroup, err = m.cluster.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if addr != "" {
|
||||
err = m.cluster.addFlashNodeToFlashGroup(addr, flashGroup)
|
||||
} else {
|
||||
err = m.cluster.selectFlashNodesFromZoneAddToFlashGroup(zoneName, count, nil, flashGroup)
|
||||
}
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView()))
|
||||
}
|
||||
|
||||
func (c *Cluster) addFlashNodeToFlashGroup(addr string, flashGroup *FlashGroup) (err error) {
|
||||
var flashNode *FlashNode
|
||||
if flashNode, err = c.setFlashNodeToFlashGroup(addr, flashGroup.ID); err != nil {
|
||||
return
|
||||
}
|
||||
flashGroup.putFlashNode(flashNode)
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) setFlashNodeToFlashGroup(addr string, flashGroupID uint64) (flashNode *FlashNode, err error) {
|
||||
if flashNode, err = c.peekFlashNode(addr); err != nil {
|
||||
return
|
||||
}
|
||||
flashNode.Lock()
|
||||
defer flashNode.Unlock()
|
||||
if !flashNode.isFlashNodeUnused() {
|
||||
err = fmt.Errorf("flashNode[%v] FlashGroupID[%v] can not add to flash group:%v", flashNode.Addr, flashNode.FlashGroupID, flashGroupID)
|
||||
return
|
||||
}
|
||||
if time.Since(flashNode.ReportTime) > time.Second*time.Duration(defaultNodeTimeOutSec) {
|
||||
flashNode.IsActive = false
|
||||
err = fmt.Errorf("flashNode[%v] is inactive lastReportTime:%v", flashNode.Addr, flashNode.ReportTime)
|
||||
return
|
||||
}
|
||||
oldFgID := flashNode.FlashGroupID
|
||||
flashNode.FlashGroupID = flashGroupID
|
||||
if err = c.syncUpdateFlashNode(flashNode); err != nil {
|
||||
flashNode.FlashGroupID = oldFgID
|
||||
return
|
||||
}
|
||||
log.LogInfo(fmt.Sprintf("action[setFlashNodeToFlashGroup] add flash node:%v to flashGroup:%v success", addr, flashGroupID))
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) selectFlashNodesFromZoneAddToFlashGroup(zoneName string, count int, excludeHosts []string, flashGroup *FlashGroup) (err error) {
|
||||
flashNodeZone, err := c.flashNodeTopo.getZone(zoneName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
flashNodeZone.mu.Lock()
|
||||
defer flashNodeZone.mu.Unlock()
|
||||
newHosts, err := flashNodeZone.selectFlashNodes(count, excludeHosts)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
successHost := make([]string, 0)
|
||||
for _, newHost := range newHosts {
|
||||
if err = c.addFlashNodeToFlashGroup(newHost, flashGroup); err != nil {
|
||||
err = fmt.Errorf("successHost:%v, newHosts:%v err:%v", successHost, newHosts, err)
|
||||
return
|
||||
}
|
||||
successHost = append(successHost, newHost)
|
||||
}
|
||||
log.LogInfo(fmt.Sprintf("action[selectFlashNodesFromZoneAddToFlashGroup] flashGroup:%v successHost:%v", flashGroup.ID, successHost))
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) flashGroupRemoveFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupNodeRemove))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupNodeRemove, metric, err, nil)
|
||||
}()
|
||||
flashGroupID, addr, zoneName, count, err := parseRequestForManageFlashNodeOfFlashGroup(r)
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
var flashGroup *FlashGroup
|
||||
if flashGroup, err = m.cluster.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if addr != "" {
|
||||
err = m.cluster.removeFlashNodeFromFlashGroup(addr, flashGroup)
|
||||
} else {
|
||||
err = m.cluster.removeFlashNodesFromTargetZone(zoneName, count, flashGroup)
|
||||
}
|
||||
if err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView()))
|
||||
}
|
||||
|
||||
func (c *Cluster) removeFlashNodeFromFlashGroup(addr string, flashGroup *FlashGroup) (err error) {
|
||||
var flashNode *FlashNode
|
||||
if flashNode, err = c.setFlashNodeToUnused(addr, flashGroup.ID); err != nil {
|
||||
return
|
||||
}
|
||||
flashGroup.removeFlashNode(flashNode.Addr)
|
||||
log.LogInfo(fmt.Sprintf("action[removeFlashNodeFromFlashGroup] node:%v flashGroup:%v, success", flashNode.Addr, flashGroup.ID))
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) removeFlashNodesFromTargetZone(zoneName string, count int, flashGroup *FlashGroup) (err error) {
|
||||
flashNodeHosts := flashGroup.getTargetZoneFlashNodeHosts(zoneName)
|
||||
if len(flashNodeHosts) < count {
|
||||
return fmt.Errorf("flashNodeHostsCount:%v less than expectCount:%v,flashNodeHosts:%v", len(flashNodeHosts), count, flashNodeHosts)
|
||||
}
|
||||
successHost := make([]string, 0)
|
||||
for _, flashNodeHost := range flashNodeHosts {
|
||||
if err = c.removeFlashNodeFromFlashGroup(flashNodeHost, flashGroup); err != nil {
|
||||
err = fmt.Errorf("successHost:%v, flashNodeHosts:%v err:%v", successHost, flashNodeHosts, err)
|
||||
return
|
||||
}
|
||||
successHost = append(successHost, flashNodeHost)
|
||||
if len(successHost) >= count {
|
||||
break
|
||||
}
|
||||
}
|
||||
log.LogInfo(fmt.Sprintf("action[removeFlashNodesFromTargetZone] flashGroup:%v successHost:%v", flashGroup.ID, successHost))
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) setFlashNodeToUnused(addr string, flashGroupID uint64) (flashNode *FlashNode, err error) {
|
||||
if flashNode, err = c.peekFlashNode(addr); err != nil {
|
||||
return
|
||||
}
|
||||
flashNode.Lock()
|
||||
defer flashNode.Unlock()
|
||||
if flashNode.FlashGroupID != flashGroupID {
|
||||
err = fmt.Errorf("flashNode[%v] FlashGroupID[%v] not equal to target flash group:%v", flashNode.Addr, flashNode.FlashGroupID, flashGroupID)
|
||||
return
|
||||
}
|
||||
oldFgID := flashNode.FlashGroupID
|
||||
flashNode.FlashGroupID = unusedFlashNodeFlashGroupID
|
||||
if err = c.syncUpdateFlashNode(flashNode); err != nil {
|
||||
flashNode.FlashGroupID = oldFgID
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) listFlashGroups(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
fgStatus proto.FlashGroupStatus
|
||||
allStatus bool
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminFlashGroupList))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.AdminFlashGroupList, metric, err, nil)
|
||||
}()
|
||||
if fgStatus, err = extractFlashGroupStatus(r); err != nil {
|
||||
if value := r.FormValue(enableKey); value == "" {
|
||||
allStatus = true // resp all flash groups
|
||||
} else {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
fgv := m.cluster.flashNodeTopo.getFlashGroupsAdminView(fgStatus, allStatus)
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fgv))
|
||||
}
|
||||
|
||||
func (m *Server) clientFlashGroups(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
flashGroupRespCache []byte
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.ClientFlashGroups))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.ClientFlashGroups, metric, err, nil)
|
||||
}()
|
||||
flashGroupRespCache, err = m.cluster.getFlashGroupResponseCache()
|
||||
if len(flashGroupRespCache) != 0 {
|
||||
send(w, r, flashGroupRespCache)
|
||||
} else {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cluster) getFlashGroupResponseCache() (flashGroupRespCache []byte, err error) {
|
||||
if cached := c.flashGroupRespCache.Load().([]byte); len(cached) == 0 {
|
||||
c.updateFlashGroupResponseCache()
|
||||
}
|
||||
flashGroupRespCache = c.flashGroupRespCache.Load().([]byte)
|
||||
if len(flashGroupRespCache) == 0 {
|
||||
return nil, fmt.Errorf("flash group resp cache is empty")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) scheduleToUpdateFlashGroupRespCache() {
|
||||
go func() {
|
||||
for {
|
||||
if c.partition != nil && c.partition.IsRaftLeader() {
|
||||
c.updateFlashGroupResponseCache()
|
||||
}
|
||||
time.Sleep(time.Second * time.Duration(c.cfg.IntervalToCheckDataPartition))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (c *Cluster) updateFlashGroupResponseCache() {
|
||||
fgv := c.flashNodeTopo.getFlashGroupView()
|
||||
reply := newSuccessHTTPReply(fgv)
|
||||
flashGroupRespCache, err := json.Marshal(reply)
|
||||
if err != nil {
|
||||
msg := fmt.Sprintf("action[updateFlashGroupResponseCache] json marshal err:%v", err)
|
||||
log.LogError(msg)
|
||||
return
|
||||
}
|
||||
c.flashGroupRespCache.Store(flashGroupRespCache)
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) clearFlashGroupResponseCache() {
|
||||
c.flashGroupRespCache.Store([]byte(nil))
|
||||
}
|
||||
|
||||
func parseRequestForManageFlashNodeOfFlashGroup(r *http.Request) (flashGroupID uint64, addr, zoneName string, count int, err error) {
|
||||
if flashGroupID, err = extractFlashGroupID(r); err != nil {
|
||||
return
|
||||
}
|
||||
if addr = r.FormValue(addrKey); addr != "" {
|
||||
return
|
||||
}
|
||||
|
||||
if zoneName, err = extractZoneName(r); err != nil {
|
||||
return
|
||||
}
|
||||
if count, err = extractCount(r); err != nil {
|
||||
return
|
||||
}
|
||||
if count <= 0 {
|
||||
err = unmatchedKey(countKey)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func parseRequestForSetFlashGroup(r *http.Request) (flashGroupID uint64, fgStatus proto.FlashGroupStatus, err error) {
|
||||
if flashGroupID, err = extractFlashGroupID(r); err != nil {
|
||||
return
|
||||
}
|
||||
fgStatus, err = extractFlashGroupStatus(r)
|
||||
return
|
||||
}
|
||||
|
||||
func extractFlashGroupStatus(r *http.Request) (fgStatus proto.FlashGroupStatus, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
var status bool
|
||||
if status, err = extractStatus(r); err != nil {
|
||||
return
|
||||
}
|
||||
if status {
|
||||
fgStatus = proto.FlashGroupStatus_Active
|
||||
} else {
|
||||
fgStatus = proto.FlashGroupStatus_Inactive
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func extractFlashGroupID(r *http.Request) (ID uint64, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
var value string
|
||||
if value = r.FormValue(idKey); value == "" {
|
||||
err = keyNotFound(idKey)
|
||||
return
|
||||
}
|
||||
return strconv.ParseUint(value, 10, 64)
|
||||
}
|
||||
|
||||
func getSetSlots(r *http.Request) (slots []uint32) {
|
||||
slots = make([]uint32, 0)
|
||||
slotStr := r.FormValue(fgSlotsKey)
|
||||
if slotStr != "" {
|
||||
arr := strings.Split(slotStr, ",")
|
||||
for i := 0; i < len(arr); i++ {
|
||||
slot, err := strconv.ParseUint(arr[i], 10, 32)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if len(slots) >= defaultFlashGroupSlotsCount {
|
||||
continue
|
||||
}
|
||||
|
||||
slots = append(slots, uint32(slot))
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func extractZoneName(r *http.Request) (name string, err error) {
|
||||
if name = r.FormValue(zoneNameKey); name == "" {
|
||||
err = keyNotFound(zoneNameKey)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func extractCount(r *http.Request) (count int, err error) {
|
||||
var value string
|
||||
if value = r.FormValue(countKey); value == "" {
|
||||
return
|
||||
}
|
||||
if count, err = strconv.Atoi(value); err != nil {
|
||||
err = unmatchedKey(countKey)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
405
master/flash_node.go
Normal file
405
master/flash_node.go
Normal file
@ -0,0 +1,405 @@
|
||||
// Copyright 2018 The CubeFS Authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied. See the License for the specific language governing
|
||||
// permissions and limitations under the License.
|
||||
|
||||
package master
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
type flashNodeValue struct {
|
||||
ID uint64
|
||||
Addr string
|
||||
ZoneName string
|
||||
Version string
|
||||
FlashGroupID uint64 // 0: have not allocated to flash group
|
||||
IsEnable bool
|
||||
}
|
||||
|
||||
type FlashNode struct {
|
||||
flashNodeValue
|
||||
sync.RWMutex
|
||||
ReportTime time.Time
|
||||
IsActive bool
|
||||
TaskManager *AdminTaskManager
|
||||
}
|
||||
|
||||
func newFlashNode(addr, zoneName, clusterID, version string, isEnable bool) *FlashNode {
|
||||
node := new(FlashNode)
|
||||
node.Addr = addr
|
||||
node.ZoneName = zoneName
|
||||
node.Version = version
|
||||
node.IsEnable = isEnable
|
||||
node.TaskManager = newAdminTaskManager(addr, clusterID)
|
||||
return node
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) isFlashNodeUnused() bool {
|
||||
return flashNode.FlashGroupID == unusedFlashNodeFlashGroupID
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) clean() {
|
||||
flashNode.TaskManager.exitCh <- struct{}{}
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) setNodeActive() {
|
||||
flashNode.Lock()
|
||||
flashNode.ReportTime = time.Now()
|
||||
flashNode.IsActive = true
|
||||
flashNode.Unlock()
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) isWriteAble() (ok bool) {
|
||||
flashNode.RLock()
|
||||
if flashNode.isFlashNodeUnused() && time.Since(flashNode.ReportTime) < time.Second*time.Duration(defaultNodeTimeOutSec) {
|
||||
ok = true
|
||||
}
|
||||
flashNode.RUnlock()
|
||||
return
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) isActiveAndEnable() bool {
|
||||
flashNode.RLock()
|
||||
defer flashNode.RUnlock()
|
||||
return flashNode.IsActive && flashNode.IsEnable
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) getFlashNodeViewInfo() (info *proto.FlashNodeViewInfo) {
|
||||
info = &proto.FlashNodeViewInfo{
|
||||
ID: flashNode.ID,
|
||||
Addr: flashNode.Addr,
|
||||
ReportTime: flashNode.ReportTime,
|
||||
IsActive: flashNode.IsActive,
|
||||
Version: flashNode.Version,
|
||||
ZoneName: flashNode.ZoneName,
|
||||
FlashGroupID: flashNode.FlashGroupID,
|
||||
IsEnable: flashNode.IsEnable,
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
|
||||
for _, t := range tasks {
|
||||
if t == nil {
|
||||
continue
|
||||
}
|
||||
node, err := c.peekFlashNode(t.OperatorAddr)
|
||||
if err != nil {
|
||||
log.LogWarn(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
|
||||
continue
|
||||
}
|
||||
if _, err = node.TaskManager.syncSendAdminTask(t); err != nil {
|
||||
log.LogError(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
|
||||
continue
|
||||
}
|
||||
node.setNodeActive()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cluster) checkFlashNodeHeartbeat() {
|
||||
tasks := make([]*proto.AdminTask, 0)
|
||||
c.flashNodeTopo.flashNodeMap.Range(func(addr, flashNode interface{}) bool {
|
||||
node := flashNode.(*FlashNode)
|
||||
node.checkLiveliness()
|
||||
task := node.createHeartbeatTask(c.masterAddr())
|
||||
tasks = append(tasks, task)
|
||||
return true
|
||||
})
|
||||
go c.syncFlashNodeHeartbeatTasks(tasks)
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) checkLiveliness() {
|
||||
flashNode.Lock()
|
||||
if time.Since(flashNode.ReportTime) > time.Second*time.Duration(defaultNodeTimeOutSec) {
|
||||
flashNode.IsActive = false
|
||||
}
|
||||
flashNode.Unlock()
|
||||
}
|
||||
|
||||
func (flashNode *FlashNode) createHeartbeatTask(masterAddr string) (task *proto.AdminTask) {
|
||||
request := &proto.HeartBeatRequest{
|
||||
CurrTime: time.Now().Unix(),
|
||||
MasterAddr: masterAddr,
|
||||
}
|
||||
task = proto.NewAdminTask(proto.OpFlashNodeHeartbeat, flashNode.Addr, request)
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) addFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
nodeAddr string
|
||||
zoneName string
|
||||
version string
|
||||
id uint64
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.FlashNodeAdd))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.FlashNodeAdd, metric, err, nil)
|
||||
}()
|
||||
if nodeAddr, zoneName, version, err = parseRequestForAddFlashNode(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
if id, err = m.cluster.addFlashNode(nodeAddr, zoneName, version); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(id))
|
||||
}
|
||||
|
||||
func (c *Cluster) addFlashNode(nodeAddr, zoneName, version string) (id uint64, err error) {
|
||||
c.flashNodeTopo.mu.Lock()
|
||||
defer func() {
|
||||
c.flashNodeTopo.mu.Unlock()
|
||||
if err != nil {
|
||||
log.LogErrorf("action[addFlashNode],clusterID[%v] Addr:%v err:%v ", c.Name, nodeAddr, err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
// TODO: update when zone changed.
|
||||
|
||||
var flashNode *FlashNode
|
||||
flashNode, err = c.peekFlashNode(nodeAddr)
|
||||
if err == nil {
|
||||
return flashNode.ID, nil
|
||||
}
|
||||
flashNode = newFlashNode(nodeAddr, zoneName, c.Name, version, true)
|
||||
_, err = c.flashNodeTopo.getZone(zoneName)
|
||||
if err != nil {
|
||||
c.flashNodeTopo.putZoneIfAbsent(newFlashNodeZone(zoneName))
|
||||
}
|
||||
if id, err = c.idAlloc.allocateCommonID(); err != nil {
|
||||
return
|
||||
}
|
||||
flashNode.ID = id
|
||||
if err = c.syncAddFlashNode(flashNode); err != nil {
|
||||
return
|
||||
}
|
||||
flashNode.ReportTime = time.Now()
|
||||
flashNode.IsActive = true
|
||||
c.flashNodeTopo.putFlashNode(flashNode)
|
||||
log.LogInfof("action[addFlashNode],clusterID[%v] Addr:%v ZoneName:%v success", c.Name, nodeAddr, zoneName)
|
||||
return
|
||||
}
|
||||
|
||||
func (m *Server) listFlashNodes(w http.ResponseWriter, r *http.Request) {
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.FlashNodeList))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.FlashNodeList, metric, nil, nil)
|
||||
}()
|
||||
zoneFlashNodes := make(map[string][]*proto.FlashNodeViewInfo)
|
||||
listAll := extractListAll(r)
|
||||
m.cluster.flashNodeTopo.flashNodeMap.Range(func(key, value interface{}) bool {
|
||||
flashNode := value.(*FlashNode)
|
||||
if listAll || flashNode.isActiveAndEnable() {
|
||||
zoneFlashNodes[flashNode.ZoneName] = append(zoneFlashNodes[flashNode.ZoneName], flashNode.getFlashNodeViewInfo())
|
||||
}
|
||||
return true
|
||||
})
|
||||
sendOkReply(w, r, newSuccessHTTPReply(zoneFlashNodes))
|
||||
}
|
||||
|
||||
func (m *Server) getFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.FlashNodeGet))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.FlashNodeGet, metric, err, nil)
|
||||
}()
|
||||
var nodeAddr string
|
||||
if nodeAddr, err = parseAndExtractNodeAddr(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
var flashNode *FlashNode
|
||||
if flashNode, err = m.cluster.peekFlashNode(nodeAddr); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(flashNode.getFlashNodeViewInfo()))
|
||||
}
|
||||
|
||||
func (m *Server) removeFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var err error
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.FlashNodeRemove))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.FlashNodeRemove, metric, err, nil)
|
||||
}()
|
||||
var offLineAddr string
|
||||
if offLineAddr, err = parseAndExtractNodeAddr(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
var node *FlashNode
|
||||
if node, err = m.cluster.peekFlashNode(offLineAddr); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(proto.ErrDataNodeNotExists))
|
||||
return
|
||||
}
|
||||
if err = m.cluster.removeFlashNode(node); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("delete flash node [%v] successfully", offLineAddr)))
|
||||
}
|
||||
|
||||
func (c *Cluster) removeFlashNode(flashNode *FlashNode) (err error) {
|
||||
log.LogWarnf("action[removeFlashNode], Node[%v] FlashGroupID:%v ZoneName:%v OffLine", flashNode.Addr, flashNode.FlashGroupID, flashNode.ZoneName)
|
||||
flashGroupID := flashNode.FlashGroupID
|
||||
if err = c.deleteFlashNode(flashNode); err != nil {
|
||||
return
|
||||
}
|
||||
if flashGroupID != unusedFlashNodeFlashGroupID {
|
||||
var flashGroup *FlashGroup
|
||||
if flashGroup, err = c.flashNodeTopo.getFlashGroup(flashGroupID); err != nil {
|
||||
return
|
||||
}
|
||||
flashGroup.removeFlashNode(flashNode.Addr)
|
||||
err = c.selectFlashNodesFromZoneAddToFlashGroup(flashNode.ZoneName, 1, []string{flashNode.Addr}, flashGroup)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
log.LogInfof("action[removeFlashNode],clusterID[%v] node[%v] flashGroupID[%v] offLine success",
|
||||
c.Name, flashNode.Addr, flashGroupID)
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) deleteFlashNode(flashNode *FlashNode) (err error) {
|
||||
flashNode.Lock()
|
||||
defer flashNode.Unlock()
|
||||
oldFlashGroupID := flashNode.FlashGroupID
|
||||
flashNode.FlashGroupID = unusedFlashNodeFlashGroupID
|
||||
if err = c.syncDeleteFlashNode(flashNode); err != nil {
|
||||
log.LogErrorf("action[deleteFlashNode],clusterID[%v] node[%v] offline failed,err[%v]",
|
||||
c.Name, flashNode.Addr, err)
|
||||
flashNode.FlashGroupID = oldFlashGroupID
|
||||
return
|
||||
}
|
||||
c.delFlashNodeFromCache(flashNode)
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) delFlashNodeFromCache(flashNode *FlashNode) {
|
||||
c.flashNodeTopo.deleteFlashNode(flashNode)
|
||||
go flashNode.clean()
|
||||
}
|
||||
|
||||
func (c *Cluster) syncAddFlashNode(flashNode *FlashNode) (err error) {
|
||||
return c.syncPutFlashNodeInfo(opSyncAddFlashNode, flashNode)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncUpdateFlashNode(flashNode *FlashNode) (err error) {
|
||||
return c.syncPutFlashNodeInfo(opSyncUpdateFlashNode, flashNode)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncDeleteFlashNode(flashNode *FlashNode) (err error) {
|
||||
return c.syncPutFlashNodeInfo(opSyncDeleteFlashNode, flashNode)
|
||||
}
|
||||
|
||||
func (c *Cluster) syncPutFlashNodeInfo(opType uint32, flashNode *FlashNode) (err error) {
|
||||
metadata := new(RaftCmd)
|
||||
metadata.Op = opType
|
||||
metadata.K = flashNodePrefix + strconv.FormatUint(flashNode.ID, 10) + keySeparator + flashNode.Addr
|
||||
metadata.V, err = json.Marshal(flashNode.flashNodeValue)
|
||||
if err != nil {
|
||||
return errors.New(err.Error())
|
||||
}
|
||||
return c.submit(metadata)
|
||||
}
|
||||
|
||||
func (c *Cluster) peekFlashNode(addr string) (flashNode *FlashNode, err error) {
|
||||
value, ok := c.flashNodeTopo.flashNodeMap.Load(addr)
|
||||
if !ok {
|
||||
err = errors.Trace(flashNodeNotFound(addr), "%v not found", addr)
|
||||
return
|
||||
}
|
||||
flashNode = value.(*FlashNode)
|
||||
return
|
||||
}
|
||||
|
||||
func flashNodeNotFound(addr string) (err error) {
|
||||
return notFoundMsg(fmt.Sprintf("flash node[%v]", addr))
|
||||
}
|
||||
|
||||
func (m *Server) setFlashNode(w http.ResponseWriter, r *http.Request) {
|
||||
var (
|
||||
nodeAddr string
|
||||
state bool
|
||||
flashNode *FlashNode
|
||||
err error
|
||||
)
|
||||
metric := exporter.NewTPCnt(apiToMetricsName(proto.FlashNodeSet))
|
||||
defer func() {
|
||||
doStatAndMetric(proto.FlashNodeSet, metric, err, nil)
|
||||
}()
|
||||
if nodeAddr, state, err = parseAndExtractFlashNode(r); err != nil {
|
||||
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
if flashNode, err = m.cluster.peekFlashNode(nodeAddr); err != nil {
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
if flashNode.IsEnable != state {
|
||||
oldState := flashNode.IsEnable
|
||||
flashNode.IsEnable = state
|
||||
if err = m.cluster.syncUpdateFlashNode(flashNode); err != nil {
|
||||
flashNode.IsEnable = oldState
|
||||
sendErrReply(w, r, newErrHTTPReply(err))
|
||||
return
|
||||
}
|
||||
}
|
||||
sendOkReply(w, r, newSuccessHTTPReply("set flashNode success"))
|
||||
}
|
||||
|
||||
func parseRequestForAddFlashNode(r *http.Request) (nodeAddr, zoneName, version string, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
if nodeAddr, err = extractNodeAddr(r); err != nil {
|
||||
return
|
||||
}
|
||||
if zoneName = r.FormValue(zoneNameKey); zoneName == "" {
|
||||
zoneName = DefaultZoneName
|
||||
}
|
||||
version = r.FormValue(versionKey)
|
||||
return
|
||||
}
|
||||
|
||||
func parseAndExtractFlashNode(r *http.Request) (nodeAddr string, state bool, err error) {
|
||||
if err = r.ParseForm(); err != nil {
|
||||
return
|
||||
}
|
||||
nodeAddr, err = extractNodeAddr(r)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
state, err = strconv.ParseBool(r.FormValue(stateKey))
|
||||
return
|
||||
}
|
||||
|
||||
func extractListAll(r *http.Request) (all bool) {
|
||||
all, _ = strconv.ParseBool(r.FormValue("all"))
|
||||
return
|
||||
}
|
||||
305
master/flash_node_topology.go
Normal file
305
master/flash_node_topology.go
Normal file
@ -0,0 +1,305 @@
|
||||
// Copyright 2018 The CubeFS Authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied. See the License for the specific language governing
|
||||
// permissions and limitations under the License.
|
||||
|
||||
package master
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
"sync"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type flashNodeTopology struct {
|
||||
mu sync.RWMutex
|
||||
|
||||
createFlashGroupLock sync.RWMutex // create/delete flashGroup
|
||||
slotsMap map[uint32]uint64 // key:slot, value: FlashGroupID
|
||||
|
||||
flashGroupMap sync.Map // key: FlashGroupID, value: *FlashGroup
|
||||
flashNodeMap sync.Map // key: FlashNodeAddr, value: *FlashNode
|
||||
zoneMap sync.Map // key: zoneName, value: *FlashNodeZone
|
||||
}
|
||||
|
||||
type FlashNodeZone struct {
|
||||
mu sync.RWMutex
|
||||
name string
|
||||
flashNode sync.Map // key: FlashNodeAddr, value: *FlashNode
|
||||
}
|
||||
|
||||
func (zone *FlashNodeZone) putFlashNode(flashNode *FlashNode) {
|
||||
zone.flashNode.Store(flashNode.Addr, flashNode)
|
||||
}
|
||||
|
||||
// the function caller should use lock of FlashNodeZone
|
||||
func (zone *FlashNodeZone) selectFlashNodes(count int, excludeHosts []string) (newHosts []string, err error) {
|
||||
zone.flashNode.Range(func(_, value interface{}) bool {
|
||||
flashNode := value.(*FlashNode)
|
||||
if contains(excludeHosts, flashNode.Addr) {
|
||||
return true
|
||||
}
|
||||
if flashNode.isWriteAble() {
|
||||
newHosts = append(newHosts, flashNode.Addr)
|
||||
}
|
||||
if len(newHosts) >= count {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
})
|
||||
if len(newHosts) != count {
|
||||
return nil, fmt.Errorf("expect count:%v newHostsCount:%v,detail:%v,excludeHosts:%v", count, len(newHosts), newHosts, excludeHosts)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func newFlashNodeZone(name string) (zone *FlashNodeZone) {
|
||||
return &FlashNodeZone{name: name}
|
||||
}
|
||||
|
||||
func newFlashNodeTopology() (t *flashNodeTopology) {
|
||||
return &flashNodeTopology{slotsMap: make(map[uint32]uint64)}
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) clear() {
|
||||
t.flashGroupMap.Range(func(key, _ interface{}) bool {
|
||||
t.flashGroupMap.Delete(key)
|
||||
return true
|
||||
})
|
||||
t.zoneMap.Range(func(key, _ interface{}) bool {
|
||||
t.zoneMap.Delete(key)
|
||||
return true
|
||||
})
|
||||
t.flashNodeMap.Range(func(key, _ interface{}) bool {
|
||||
t.flashNodeMap.Delete(key)
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) getZone(name string) (zone *FlashNodeZone, err error) {
|
||||
if name == "" {
|
||||
return nil, fmt.Errorf("zone name is empty")
|
||||
}
|
||||
value, ok := t.zoneMap.Load(name)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("zone[%s] not found", name)
|
||||
}
|
||||
if zone = value.(*FlashNodeZone); zone == nil {
|
||||
return nil, fmt.Errorf("zone[%s] not found", name)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) putZoneIfAbsent(zone *FlashNodeZone) (old *FlashNodeZone) {
|
||||
oldZone, loaded := t.zoneMap.LoadOrStore(zone.name, zone)
|
||||
if loaded {
|
||||
return oldZone.(*FlashNodeZone)
|
||||
}
|
||||
return zone
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) putFlashNode(flashNode *FlashNode) (err error) {
|
||||
if _, loaded := t.flashNodeMap.LoadOrStore(flashNode.Addr, flashNode); loaded {
|
||||
return
|
||||
}
|
||||
zone, err := t.getZone(flashNode.ZoneName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
zone.putFlashNode(flashNode)
|
||||
return
|
||||
}
|
||||
|
||||
// TODO: remove from group.
|
||||
func (t *flashNodeTopology) deleteFlashNode(flashNode *FlashNode) {
|
||||
t.flashNodeMap.Delete(flashNode.Addr)
|
||||
zone, err := t.getZone(flashNode.ZoneName)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
zone.flashNode.Delete(flashNode.Addr)
|
||||
}
|
||||
|
||||
// the function caller should use createFlashGroupLock
|
||||
func (t *flashNodeTopology) allocateNewSlotsForCreateFlashGroup(fgID uint64, setSlots []uint32) (slots []uint32) {
|
||||
slots = make([]uint32, 0, defaultFlashGroupSlotsCount)
|
||||
if len(setSlots) != 0 {
|
||||
slots = append(slots, setSlots...)
|
||||
}
|
||||
for len(slots) < defaultFlashGroupSlotsCount {
|
||||
slot := allocateNewSlot()
|
||||
if _, ok := t.slotsMap[slot]; ok {
|
||||
continue
|
||||
}
|
||||
slots = append(slots, slot)
|
||||
t.slotsMap[slot] = fgID
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func allocateNewSlot() (slot uint32) {
|
||||
bytes, _ := uuid.New().MarshalBinary()
|
||||
slot = crc32.ChecksumIEEE(bytes)
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) removeSlots(slots []uint32) {
|
||||
for _, slot := range slots {
|
||||
delete(t.slotsMap, slot)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) createFlashGroup(fgID uint64, c *Cluster, setSlots []uint32) (flashGroup *FlashGroup, err error) {
|
||||
t.createFlashGroupLock.Lock()
|
||||
defer t.createFlashGroupLock.Unlock()
|
||||
slots := t.allocateNewSlotsForCreateFlashGroup(fgID, setSlots)
|
||||
flashGroup = newFlashGroup(fgID, slots, proto.FlashGroupStatus_Inactive)
|
||||
if err = c.syncAddFlashGroup(flashGroup); err != nil {
|
||||
t.removeSlots(slots)
|
||||
return
|
||||
}
|
||||
t.flashGroupMap.Store(flashGroup.ID, flashGroup)
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) removeFlashGroup(flashGroup *FlashGroup, c *Cluster) (err error) {
|
||||
t.createFlashGroupLock.Lock()
|
||||
defer t.createFlashGroupLock.Unlock()
|
||||
slots := flashGroup.Slots
|
||||
oldStatus := flashGroup.Status
|
||||
|
||||
flashGroup.Status = proto.FlashGroupStatus_Inactive
|
||||
if err = c.syncDeleteFlashGroup(flashGroup); err != nil {
|
||||
flashGroup.Status = oldStatus
|
||||
return
|
||||
}
|
||||
t.removeSlots(slots)
|
||||
t.flashGroupMap.Delete(flashGroup.ID)
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) getFlashGroup(fgID uint64) (flashGroup *FlashGroup, err error) {
|
||||
value, ok := t.flashGroupMap.Load(fgID)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("flashGroup[%v] is not found", fgID)
|
||||
}
|
||||
flashGroup = value.(*FlashGroup)
|
||||
if flashGroup == nil {
|
||||
return nil, fmt.Errorf("flashGroup[%v] is not found", fgID)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) getFlashGroupView() (fgv *proto.FlashGroupView) {
|
||||
fgv = new(proto.FlashGroupView)
|
||||
t.flashGroupMap.Range(func(_, value interface{}) bool {
|
||||
fg := value.(*FlashGroup)
|
||||
if fg.Status.IsActive() {
|
||||
hosts := fg.getFlashNodeHosts(true)
|
||||
if len(hosts) == 0 {
|
||||
return true
|
||||
}
|
||||
fgv.FlashGroups = append(fgv.FlashGroups, &proto.FlashGroupInfo{
|
||||
ID: fg.ID,
|
||||
Slot: fg.Slots,
|
||||
Hosts: hosts,
|
||||
})
|
||||
//for _, slot := range fg.Slots {
|
||||
// fgv.FlashGroups = append(fgv.FlashGroups, proto.FlashGroupInfo{
|
||||
// ID: fg.ID,
|
||||
// Slot: slot,
|
||||
// Hosts: hosts,
|
||||
// })
|
||||
//}
|
||||
}
|
||||
return true
|
||||
})
|
||||
//sort.Slice(fgv.FlashGroups, func(i, j int) bool {
|
||||
// return fgv.FlashGroups[i].Slot < fgv.FlashGroups[j].Slot
|
||||
//})
|
||||
return
|
||||
}
|
||||
|
||||
func (t *flashNodeTopology) getFlashGroupsAdminView(fgStatus proto.FlashGroupStatus, allStatus bool) (fgv *proto.FlashGroupsAdminView) {
|
||||
fgv = new(proto.FlashGroupsAdminView)
|
||||
t.flashGroupMap.Range(func(_, value interface{}) bool {
|
||||
fg := value.(*FlashGroup)
|
||||
if allStatus || fg.Status == fgStatus {
|
||||
fgv.FlashGroups = append(fgv.FlashGroups, fg.GetAdminView())
|
||||
}
|
||||
return true
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) loadFlashNodes() (err error) {
|
||||
result, err := c.fsm.store.SeekForPrefix([]byte(flashNodePrefix))
|
||||
if err != nil {
|
||||
err = fmt.Errorf("action[loadFlashNodes],err:%v", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
for _, value := range result {
|
||||
fnv := &flashNodeValue{}
|
||||
if err = json.Unmarshal(value, fnv); err != nil {
|
||||
err = fmt.Errorf("action[loadFlashNodes],value:%v,unmarshal err:%v", string(value), err)
|
||||
return
|
||||
}
|
||||
flashNode := newFlashNode(fnv.Addr, fnv.ZoneName, c.Name, fnv.Version, fnv.IsEnable)
|
||||
flashNode.ID = fnv.ID
|
||||
flashNode.FlashGroupID = fnv.FlashGroupID
|
||||
|
||||
if !flashNode.isFlashNodeUnused() {
|
||||
if flashGroup, err1 := c.flashNodeTopo.getFlashGroup(flashNode.FlashGroupID); err1 == nil {
|
||||
flashGroup.putFlashNode(flashNode)
|
||||
} else {
|
||||
log.LogErrorf("action[loadFlashNodes]fnv:%v err:%v", *fnv, err1.Error())
|
||||
}
|
||||
}
|
||||
|
||||
_, err = c.flashNodeTopo.getZone(flashNode.ZoneName)
|
||||
if err != nil {
|
||||
c.flashNodeTopo.putZoneIfAbsent(newFlashNodeZone(flashNode.ZoneName))
|
||||
}
|
||||
c.flashNodeTopo.putFlashNode(flashNode)
|
||||
log.LogInfof("action[loadFlashNodes],flashNode[%v],FlashGroupID[%v]", flashNode.Addr, flashNode.FlashGroupID)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Cluster) loadFlashGroups() (err error) {
|
||||
result, err := c.fsm.store.SeekForPrefix([]byte(flashGroupPrefix))
|
||||
if err != nil {
|
||||
err = fmt.Errorf("action[loadFlashGroups],err:%v", err.Error())
|
||||
return err
|
||||
}
|
||||
for _, value := range result {
|
||||
var fgv flashGroupValue
|
||||
if err = json.Unmarshal(value, &fgv); err != nil {
|
||||
err = fmt.Errorf("action[loadFlashGroups],value:%v,unmarshal err:%v", string(value), err)
|
||||
return
|
||||
}
|
||||
flashGroup := newFlashGroup(fgv.ID, fgv.Slots, fgv.Status)
|
||||
c.flashNodeTopo.flashGroupMap.Store(flashGroup.ID, flashGroup)
|
||||
for _, slot := range flashGroup.Slots {
|
||||
c.flashNodeTopo.slotsMap[slot] = flashGroup.ID
|
||||
}
|
||||
log.LogInfof("action[loadFlashGroups],flashGroup[%v]", flashGroup.ID)
|
||||
}
|
||||
return
|
||||
}
|
||||
@ -850,6 +850,23 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
|
||||
router.NewRoute().Methods(http.MethodDelete, http.MethodPost).
|
||||
Path(proto.S3QoSDelete).
|
||||
HandlerFunc(m.S3QosDelete)
|
||||
|
||||
// APIs for FlashNode
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.FlashNodeAdd).HandlerFunc(m.addFlashNode)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.FlashNodeSet).HandlerFunc(m.setFlashNode)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.FlashNodeRemove).HandlerFunc(m.removeFlashNode)
|
||||
router.NewRoute().Methods(http.MethodGet).Path(proto.FlashNodeGet).HandlerFunc(m.getFlashNode)
|
||||
router.NewRoute().Methods(http.MethodGet).Path(proto.FlashNodeList).HandlerFunc(m.listFlashNodes)
|
||||
|
||||
// APIs for FlashGroup
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupCreate).HandlerFunc(m.createFlashGroup)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupSet).HandlerFunc(m.setFlashGroup)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupRemove).HandlerFunc(m.removeFlashGroup)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupNodeAdd).HandlerFunc(m.flashGroupAddFlashNode)
|
||||
router.NewRoute().Methods(http.MethodPost).Path(proto.AdminFlashGroupNodeRemove).HandlerFunc(m.flashGroupRemoveFlashNode)
|
||||
router.NewRoute().Methods(http.MethodGet).Path(proto.AdminFlashGroupGet).HandlerFunc(m.getFlashGroup)
|
||||
router.NewRoute().Methods(http.MethodGet).Path(proto.AdminFlashGroupList).HandlerFunc(m.listFlashGroups)
|
||||
router.NewRoute().Methods(http.MethodGet).Path(proto.ClientFlashGroups).HandlerFunc(m.clientFlashGroups)
|
||||
}
|
||||
|
||||
func (m *Server) registerHandler(router *mux.Router, model string, schema *graphql.Schema) {
|
||||
|
||||
@ -293,6 +293,10 @@ func (m *Server) clearMetadata() {
|
||||
|
||||
m.cluster.t = newTopology()
|
||||
// m.cluster.apiLimiter.Clear()
|
||||
|
||||
m.cluster.flashNodeTopo.clear()
|
||||
m.cluster.clearFlashGroupResponseCache()
|
||||
m.cluster.flashNodeTopo = newFlashNodeTopology()
|
||||
}
|
||||
|
||||
func (m *Server) refreshUser() (err error) {
|
||||
|
||||
@ -273,6 +273,22 @@ const (
|
||||
AdminEnablePersistAccessTime = "/vol/enablePersistAccessTime"
|
||||
|
||||
AdminVolAddAllowedStorageClass = "/vol/addAllowedStorageClass"
|
||||
// FlashNode API
|
||||
FlashNodeAdd = "/flashNode/add"
|
||||
FlashNodeSet = "/flashNode/set"
|
||||
FlashNodeRemove = "/flashNode/remove"
|
||||
FlashNodeGet = "/flashNode/get"
|
||||
FlashNodeList = "/flashNode/list"
|
||||
|
||||
// FlashGroup API
|
||||
AdminFlashGroupCreate = "/flashGroup/create"
|
||||
AdminFlashGroupSet = "/flashGroup/set"
|
||||
AdminFlashGroupRemove = "/flashGroup/remove"
|
||||
AdminFlashGroupNodeAdd = "/flashGroup/addFlashNode"
|
||||
AdminFlashGroupNodeRemove = "/flashGroup/removeFlashNode"
|
||||
AdminFlashGroupGet = "/flashGroup/get"
|
||||
AdminFlashGroupList = "/flashGroup/list"
|
||||
ClientFlashGroups = "/client/flashGroups"
|
||||
)
|
||||
|
||||
var GApiInfo map[string]string = map[string]string{
|
||||
|
||||
@ -184,6 +184,7 @@ const (
|
||||
ErrCodeNodeSetNotExists
|
||||
ErrCodeNoSuchLifecycleConfiguration
|
||||
ErrCodeNoSupportStorageClass
|
||||
ErrCodeTmpfsNoSpace
|
||||
)
|
||||
|
||||
// Err2CodeMap error map to code
|
||||
@ -251,6 +252,7 @@ var Err2CodeMap = map[error]int32{
|
||||
ErrNodeSetNotExists: ErrCodeNodeSetNotExists,
|
||||
ErrNoSuchLifecycleConfiguration: ErrCodeNoSuchLifecycleConfiguration,
|
||||
ErrNoSupportStorageClass: ErrCodeNoSupportStorageClass,
|
||||
ErrTmpfsNoSpace: ErrCodeTmpfsNoSpace,
|
||||
}
|
||||
|
||||
func ParseErrorCode(code int32) error {
|
||||
@ -327,6 +329,7 @@ var code2ErrMap = map[int32]error{
|
||||
ErrCodeVolHasDeleted: ErrVolHasDeleted,
|
||||
ErrCodeNoSuchLifecycleConfiguration: ErrNoSuchLifecycleConfiguration,
|
||||
ErrCodeNoSupportStorageClass: ErrNoSupportStorageClass,
|
||||
ErrCodeTmpfsNoSpace: ErrTmpfsNoSpace,
|
||||
}
|
||||
|
||||
type GeneralResp struct {
|
||||
|
||||
@ -162,6 +162,7 @@ type ClusterView struct {
|
||||
ForbidWriteOpOfProtoVer0 bool
|
||||
LegacyDataMediaType uint32
|
||||
RaftPartitionCanUsingDifferentPortEnabled bool
|
||||
FlashNodes []NodeView
|
||||
}
|
||||
|
||||
// ClusterNode defines the structure of a cluster node
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
