diff --git a/docker/conf/flashgroupmanager1.json b/docker/conf/flashgroupmanager1.json index 862738845..44c7147df 100644 --- a/docker/conf/flashgroupmanager1.json +++ b/docker/conf/flashgroupmanager1.json @@ -3,8 +3,12 @@ "id": "1", "role": "flashgroupmanager", "ip": "192.168.0.81", + "peers": "1:192.168.0.81:17010,2:192.168.0.82:17010,3:192.168.0.83:17010", + "retainLogs": "20000", "listen": "17010", "prof": "17020", "logLevel": "debug", - "logDir": "/cfs/log" + "logDir": "/cfs/log", + "walDir": "/cfs/data/wal", + "storeDir": "/cfs/data/store" } \ No newline at end of file diff --git a/docker/conf/flashgroupmanager2.json b/docker/conf/flashgroupmanager2.json new file mode 100644 index 000000000..873a4436d --- /dev/null +++ b/docker/conf/flashgroupmanager2.json @@ -0,0 +1,14 @@ +{ + "clusterName": "cubefs01", + "id": "2", + "role": "flashgroupmanager", + "ip": "192.168.0.82", + "peers": "1:192.168.0.81:17010,2:192.168.0.82:17010,3:192.168.0.83:17010", + "retainLogs": "20000", + "listen": "17010", + "prof": "17020", + "logLevel": "debug", + "logDir": "/cfs/log", + "walDir": "/cfs/data/wal", + "storeDir": "/cfs/data/store" +} \ No newline at end of file diff --git a/docker/conf/flashgroupmanager3.json b/docker/conf/flashgroupmanager3.json new file mode 100644 index 000000000..a04420033 --- /dev/null +++ b/docker/conf/flashgroupmanager3.json @@ -0,0 +1,14 @@ +{ + "clusterName": "cubefs01", + "id": "3", + "role": "flashgroupmanager", + "ip": "192.168.0.83", + "peers": "1:192.168.0.81:17010,2:192.168.0.82:17010,3:192.168.0.83:17010", + "retainLogs": "20000", + "listen": "17010", + "prof": "17020", + "logLevel": "debug", + "logDir": "/cfs/log", + "walDir": "/cfs/data/wal", + "storeDir": "/cfs/data/store" +} \ No newline at end of file diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index fc50c6fd7..9f4bfe7fd 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -113,6 +113,8 @@ services: - flashnode2 - flashnode3 - flashgroupmanager1 + - flashgroupmanager2 + - flashgroupmanager3 - console1 - nginx networks: @@ -345,7 +347,28 @@ services: networks: extnetwork: ipv4_address: 192.168.0.81 - + flashgroupmanager2: + <<: *flashgroupmanager + volumes: + - ${DiskPath:-./docker_data}/flashgroupmanager2/data:/cfs/data + - ./bin:/cfs/bin:ro + - ${DiskPath:-./docker_data}/flashgroupmanager2/log:/cfs/log + - ./conf/flashgroupmanager2.json:/cfs/conf/flashgroupmanager.json + - ./script/start_flashgroupmanager.sh:/cfs/script/start.sh + networks: + extnetwork: + ipv4_address: 192.168.0.82 + flashgroupmanager3: + <<: *flashgroupmanager + volumes: + - ${DiskPath:-./docker_data}/flashgroupmanager3/data:/cfs/data + - ./bin:/cfs/bin:ro + - ${DiskPath:-./docker_data}/flashgroupmanager3/log:/cfs/log + - ./conf/flashgroupmanager3.json:/cfs/conf/flashgroupmanager.json + - ./script/start_flashgroupmanager.sh:/cfs/script/start.sh + networks: + extnetwork: + ipv4_address: 192.168.0.83 console1: <<: *servernode ports: diff --git a/remotecache/flashgroupmanager/api_args_parse.go b/remotecache/flashgroupmanager/api_args_parse.go index 347f3157f..c27831fac 100644 --- a/remotecache/flashgroupmanager/api_args_parse.go +++ b/remotecache/flashgroupmanager/api_args_parse.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "strconv" + "strings" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/util/compressor" @@ -87,3 +88,28 @@ func parseRequestToGetTaskResponse(r *http.Request) (tr *proto.AdminTask, err er err = decoder.Decode(tr) return } + +func parseRequestForRaftNode(r *http.Request) (id uint64, host string, err error) { + if err = r.ParseForm(); err != nil { + return + } + var idStr string + if idStr = r.FormValue(idKey); idStr == "" { + err = keyNotFound(idKey) + return + } + + if id, err = strconv.ParseUint(idStr, 10, 64); err != nil { + return + } + if host = r.FormValue(addrKey); host == "" { + err = keyNotFound(addrKey) + return + } + + if arr := strings.Split(host, colonSplit); len(arr) < 2 { + err = unmatchedKey(addrKey) + return + } + return +} diff --git a/remotecache/flashgroupmanager/api_service.go b/remotecache/flashgroupmanager/api_service.go index 73236ff38..a0aae6bad 100644 --- a/remotecache/flashgroupmanager/api_service.go +++ b/remotecache/flashgroupmanager/api_service.go @@ -2,14 +2,16 @@ package flashgroupmanager import ( "fmt" - "github.com/cubefs/cubefs/cmd/common" - "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/log" "net/http" "strconv" "strings" "time" + "github.com/cubefs/cubefs/cmd/common" "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/auditlog" "github.com/cubefs/cubefs/util/exporter" "github.com/cubefs/cubefs/util/iputil" "github.com/cubefs/cubefs/util/stat" @@ -52,17 +54,17 @@ func (m *FlashGroupManager) getCluster(w http.ResponseWriter, r *http.Request) { sendErrReply(w, r, newErrHTTPReply(err)) return } - //TODO cv := &proto.ClusterView{ - Name: m.cluster.Name, - CreateTime: time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat), - //LeaderAddr: m.leaderInfo.addr, - //FlashNodes: make([]proto.NodeView, 0), - //FlashNodeHandleReadTimeout: m.cluster.cfg.flashNodeHandleReadTimeout, - //FlashNodeReadDataNodeTimeout: m.cluster.cfg.flashNodeReadDataNodeTimeout, + Name: m.cluster.Name, + CreateTime: time.Unix(m.cluster.CreateTime, 0).Format(proto.TimeFormat), + LeaderAddr: m.leaderInfo.addr, + FlashNodes: make([]proto.NodeView, 0), + FlashNodeHandleReadTimeout: m.cluster.cfg.flashNodeHandleReadTimeout, + FlashNodeReadDataNodeTimeout: m.cluster.cfg.flashNodeReadDataNodeTimeout, } cv.DataNodeStatInfo = new(proto.NodeStatInfo) cv.MetaNodeStatInfo = new(proto.NodeStatInfo) + cv.FlashNodes = m.cluster.allFlashNodes() sendOkReply(w, r, newSuccessHTTPReply(cv)) } @@ -96,11 +98,11 @@ func (m *FlashGroupManager) clientFlashGroups(w http.ResponseWriter, r *http.Req doStatAndMetric(proto.ClientFlashGroups, metric, err, nil) }() - //TODO - //if !m.metaReady { - // sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("meta not ready"))) - // return - //} + if !m.metaReady { + sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("meta not ready"))) + return + } + cache := m.cluster.flashNodeTopo.getClientResponse() if len(cache) == 0 { sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("flash group response cache is empty"))) @@ -210,13 +212,12 @@ func (m *FlashGroupManager) setFlashGroup(w http.ResponseWriter, r *http.Request oldStatus := flashGroup.Status flashGroup.Status = fgStatus if oldStatus != fgStatus { - // TODO - //if err = m.cluster.syncUpdateFlashGroup(flashGroup); err != nil { - // flashGroup.Status = oldStatus - // flashGroup.lock.Unlock() - // sendErrReply(w, r, newErrHTTPReply(err)) - // return - //} + if err = m.cluster.syncUpdateFlashGroup(flashGroup); err != nil { + flashGroup.Status = oldStatus + flashGroup.lock.Unlock() + sendErrReply(w, r, newErrHTTPReply(err)) + return + } m.cluster.flashNodeTopo.updateClientCache() } flashGroup.lock.Unlock() @@ -573,6 +574,88 @@ func (m *FlashGroupManager) flashGroupRemoveFlashNode(w http.ResponseWriter, r * sendOkReply(w, r, newSuccessHTTPReply(flashGroup.GetAdminView())) } +func (m *FlashGroupManager) addRaftNode(w http.ResponseWriter, r *http.Request) { + var ( + id uint64 + addr string + err error + ) + metric := exporter.NewTPCnt(apiToMetricsName(proto.AddRaftNode)) + defer func() { + doStatAndMetric(proto.AddRaftNode, metric, err, nil) + AuditLog(r, proto.AddRaftNode, fmt.Sprintf("add raft node id :%v, addr:%v", id, addr), err) + }() + + var msg string + id, addr, err = parseRequestForRaftNode(r) + if err != nil { + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()}) + return + } + + if err = m.cluster.addRaftNode(id, addr); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + msg = fmt.Sprintf("add raft node id :%v, addr:%v successfully \n", id, addr) + sendOkReply(w, r, newSuccessHTTPReply(msg)) +} + +func (m *FlashGroupManager) removeRaftNode(w http.ResponseWriter, r *http.Request) { + var ( + id uint64 + addr string + err error + ) + metric := exporter.NewTPCnt(apiToMetricsName(proto.RemoveRaftNode)) + defer func() { + doStatAndMetric(proto.RemoveRaftNode, metric, err, nil) + AuditLog(r, proto.RemoveRaftNode, fmt.Sprintf("del raft node id :%v, addr:%v", id, addr), err) + }() + + var msg string + id, addr, err = parseRequestForRaftNode(r) + if err != nil { + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()}) + return + } + err = m.cluster.removeRaftNode(id, addr) + if err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + msg = fmt.Sprintf("remove raft node id :%v,adr:%v successfully\n", id, addr) + sendOkReply(w, r, newSuccessHTTPReply(msg)) +} + +func (m *FlashGroupManager) getRaftStatus(w http.ResponseWriter, r *http.Request) { + metric := exporter.NewTPCnt(apiToMetricsName(proto.RaftStatus)) + defer func() { + doStatAndMetric(proto.RaftStatus, metric, nil, nil) + }() + + data := m.raftStore.RaftStatus(GroupID) + log.LogInfof("get raft status, %s", data.String()) + sendOkReply(w, r, newSuccessHTTPReply(data)) +} + +func (m *FlashGroupManager) changeMasterLeader(w http.ResponseWriter, r *http.Request) { + var err error + metric := exporter.NewTPCnt(apiToMetricsName(proto.AdminChangeMasterLeader)) + defer func() { + doStatAndMetric(proto.AdminChangeMasterLeader, metric, err, nil) + AuditLog(r, proto.AdminChangeMasterLeader, "", err) + }() + + if err = m.cluster.tryToChangeLeaderByHost(); err != nil { + log.LogErrorf("changeMasterLeader.err %v", err) + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()}) + return + } + rstMsg := " changeMasterLeader. command success send to dest host but need check. " + _ = sendOkReply(w, r, newSuccessHTTPReply(rstMsg)) +} + func getSetSlots(r *http.Request) (slots []uint32, err error) { r.ParseForm() slots = make([]uint32, 0) @@ -655,3 +738,8 @@ func parseArgsFlashGroupNode(r *http.Request) (id uint64, addr, zoneName string, } return } + +func AuditLog(r *http.Request, op, msg string, err error) { + head := fmt.Sprintf("%s %s", r.RemoteAddr, op) + auditlog.LogMasterOp(head, msg, err) +} diff --git a/remotecache/flashgroupmanager/cluster.go b/remotecache/flashgroupmanager/cluster.go index 8ca935c06..7ecd5bd6b 100644 --- a/remotecache/flashgroupmanager/cluster.go +++ b/remotecache/flashgroupmanager/cluster.go @@ -1,15 +1,18 @@ package flashgroupmanager import ( + "encoding/json" "fmt" - "github.com/cubefs/cubefs/proto" - "github.com/cubefs/cubefs/util/errors" + "sort" "strconv" "strings" "sync" "time" + "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/raftstore" "github.com/cubefs/cubefs/sdk/httpclient" + "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/log" ) @@ -20,17 +23,23 @@ type Cluster struct { idAlloc *IDAllocator stopc chan bool stopFlag int32 - wg sync.WaitGroup //run task? + wg sync.WaitGroup cfg *clusterConfig + leaderInfo *LeaderInfo + fsm *MetadataFsm + partition raftstore.Partition } -func newCluster(name string, cfg *clusterConfig) (c *Cluster) { +func newCluster(name string, cfg *clusterConfig, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition raftstore.Partition) (c *Cluster) { c = new(Cluster) c.Name = name c.flashNodeTopo = NewFlashNodeTopology() c.stopc = make(chan bool) - c.idAlloc = newIDAllocator() + c.fsm = fsm + c.partition = partition c.cfg = cfg + c.leaderInfo = leaderInfo + c.idAlloc = newIDAllocator(c.fsm.store, c.partition) return } @@ -130,13 +139,12 @@ func (c *Cluster) setFlashNodeToUnused(addr string, flashGroupID uint64) (flashN return } - // oldFgID := flashNode.FlashGroupID + oldFgID := flashNode.FlashGroupID flashNode.FlashGroupID = UnusedFlashNodeFlashGroupID - // TODO - //if err = c.syncUpdateFlashNode(flashNode); err != nil { - // flashNode.FlashGroupID = oldFgID - // return - //} + if err = c.syncUpdateFlashNode(flashNode); err != nil { + flashNode.FlashGroupID = oldFgID + return + } go func() { time.Sleep(time.Duration(DefaultWaitClientUpdateFgTimeSec) * time.Second) @@ -175,10 +183,9 @@ func (c *Cluster) addFlashNode(nodeAddr, zoneName, version string) (id uint64, e return } flashNode.ID = id - //TODO - //if err = c.syncAddFlashNode(flashNode); err != nil { - // return - //} + if err = c.syncAddFlashNode(flashNode); err != nil { + return + } flashNode.ReportTime = time.Now() flashNode.IsActive = true if err = c.flashNodeTopo.putFlashNode(flashNode); err != nil { @@ -192,13 +199,12 @@ func (c *Cluster) updateFlashNode(flashNode *FlashNode, enable bool) (err error) flashNode.Lock() defer flashNode.Unlock() if flashNode.IsEnable != enable { - //oldState := flashNode.IsEnable + oldState := flashNode.IsEnable flashNode.IsEnable = enable - // TODO - //if err = c.syncUpdateFlashNode(flashNode); err != nil { - // flashNode.IsEnable = oldState - // return - //} + if err = c.syncUpdateFlashNode(flashNode); err != nil { + flashNode.IsEnable = oldState + return + } if flashNode.FlashGroupID != UnusedFlashNodeFlashGroupID { c.flashNodeTopo.updateClientCache() } @@ -210,10 +216,9 @@ func (c *Cluster) updateFlashNodeWorkRole(flashNode *FlashNode, workRole string) flashNode.Lock() defer flashNode.Unlock() flashNode.WorkRole = workRole - // TODO - //if err := c.syncUpdateFlashNode(flashNode); err != nil { - // return err - //} + if err := c.syncUpdateFlashNode(flashNode); err != nil { + return err + } return nil } @@ -253,13 +258,12 @@ func (c *Cluster) deleteFlashNode(flashNode *FlashNode) (oldFlashGroupID uint64, defer flashNode.Unlock() oldFlashGroupID = flashNode.FlashGroupID flashNode.FlashGroupID = UnusedFlashNodeFlashGroupID - //TODO - //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 - //} + 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 } @@ -293,13 +297,12 @@ func (c *Cluster) setFlashNodeToFlashGroup(addr string, flashGroupID uint64) (fl err = fmt.Errorf("flashNode[%v] is inactive lastReportTime:%v", flashNode.Addr, flashNode.ReportTime) return } - //oldFgID := flashNode.FlashGroupID + oldFgID := flashNode.FlashGroupID flashNode.FlashGroupID = flashGroupID - // TODO - //if err = c.syncUpdateFlashNode(flashNode); err != nil { - // flashNode.FlashGroupID = oldFgID - // return - //} + 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 } @@ -331,10 +334,9 @@ func (c *Cluster) scheduleToUpdateFlashGroupRespCache() { ticker := time.NewTicker(dur) defer ticker.Stop() for { - // TODO - //if c.partition != nil && c.partition.IsRaftLeader() { - // c.flashNodeTopo.updateClientResponse() - //} + if c.partition != nil && c.partition.IsRaftLeader() { + c.flashNodeTopo.updateClientResponse() + } select { case <-c.stopc: return @@ -349,6 +351,7 @@ func (c *Cluster) scheduleToUpdateFlashGroupRespCache() { func (c *Cluster) scheduleTask() { c.scheduleToUpdateFlashGroupRespCache() c.scheduleToCheckHeartbeat() + c.scheduleToUpdateFlashGroupSlots() } func (c *Cluster) peekFlashNode(addr string) (flashNode *FlashNode, err error) { @@ -381,10 +384,6 @@ func (c *Cluster) handleFlashNodeTaskResponse(nodeAddr string, task *proto.Admin } switch task.OpCode { - //TODO - //case proto.OpFlashNodeScan: - // response := task.Response.(*proto.FlashNodeManualTaskResponse) - // err = c.handleFlashNodeScanResp(task.OperatorAddr, response) case proto.OpFlashNodeHeartbeat: response := task.Response.(*proto.FlashNodeHeartbeatResponse) err = c.handleFlashNodeHeartbeatResp(task.OperatorAddr, response) @@ -415,8 +414,6 @@ func (c *Cluster) handleFlashNodeHeartbeatResp(nodeAddr string, resp *proto.Flas } node.setActive() node.updateFlashNodeStatHeartbeat(resp) - //TODO: preload - // c.handleManualTaskProcessing(node, resp) return } @@ -452,16 +449,149 @@ func (c *Cluster) scheduleToCheckHeartbeat() { tickTime: time.Second * defaultIntervalToCheckHeartbeat, name: "scheduleToCheckHeartbeat_checkFlashNodeHeartbeat", function: func() (fin bool) { - //TODO - //if c.partition != nil && c.partition.IsRaftLeader() { - // c.checkLcNodeHeartbeat() - //} - c.checkFlashNodeHeartbeat() + if c.partition != nil && c.partition.IsRaftLeader() { + c.checkFlashNodeHeartbeat() + } return }, }) } +func (c *Cluster) scheduleToUpdateFlashGroupSlots() { + c.runTask( + &cTask{ + tickTime: time.Minute, + name: "scheduleToUpdateFlashGroupSlots", + function: func() (fin bool) { + if c.partition != nil && c.partition.IsRaftLeader() { + isNotUpdated := true + c.flashNodeTopo.flashGroupMap.Range(func(key, value interface{}) bool { + flashGroup := value.(*FlashGroup) + c.flashNodeTopo.createFlashGroupLock.Lock() + defer c.flashNodeTopo.createFlashGroupLock.Unlock() + slotStatus := flashGroup.getSlotStatus() + if slotStatus == proto.SlotStatus_Creating && (!flashGroup.GetStatus().IsActive() || len(flashGroup.flashNodes) == 0) { + return true + } + if slotStatus == proto.SlotStatus_Completed { + return true + } else if slotStatus == proto.SlotStatus_Creating || slotStatus == proto.SlotStatus_Deleting { + if err := c.updateFlashGroupSlots(flashGroup); err == nil { + isNotUpdated = false + } + return true + } else { + log.LogWarnf("scheduleToUpdateFlashGroupSlots failed, flashGroup(%v) has unknown SlotStatus(%v)", flashGroup.ID, flashGroup.SlotStatus) + return true + } + }) + if !isNotUpdated { + c.flashNodeTopo.updateClientCache() + } + } + return + }, + }) +} + +func (c *Cluster) updateFlashGroupSlots(flashGroup *FlashGroup) (err error) { + var updatedSlotsNum uint32 + var newSlotStatus proto.SlotStatus + var newPendingSlots []uint32 + var needDeleteFgFlag bool + + if flashGroup.getSlotStatus() == proto.SlotStatus_Deleting { + if needDeleteFgFlag, err = c.checkShrinkOrDeleteFlashGroup(flashGroup); err != nil { + return + } + } + + flashGroup.lock.Lock() + leftPendingSlotsNum := uint32(len(flashGroup.PendingSlots)) - flashGroup.Step + oldSlots := flashGroup.Slots + oldPendingSlots := flashGroup.PendingSlots + oldSlotStatus := flashGroup.SlotStatus + oldStatus := flashGroup.Status + if leftPendingSlotsNum > 0 { // previous steps + updatedSlotsNum = flashGroup.Step + newPendingSlots = oldPendingSlots[updatedSlotsNum:] + newSlotStatus = oldSlotStatus + } else { // final step + updatedSlotsNum = uint32(len(flashGroup.PendingSlots)) + newPendingSlots = nil + newSlotStatus = proto.SlotStatus_Completed + } + newSlots := getNewSlots(flashGroup.Slots, flashGroup.PendingSlots[:updatedSlotsNum], oldSlotStatus) + flashGroup.Slots = newSlots + flashGroup.PendingSlots = newPendingSlots + flashGroup.SlotStatus = newSlotStatus + if needDeleteFgFlag { + flashGroup.Status = proto.FlashGroupStatus_Inactive + err = c.syncDeleteFlashGroup(flashGroup) + } else { + err = c.syncUpdateFlashGroup(flashGroup) + } + + if err != nil { + flashGroup.Slots = oldSlots + flashGroup.PendingSlots = oldPendingSlots + flashGroup.SlotStatus = oldSlotStatus + if needDeleteFgFlag { + flashGroup.Status = oldStatus + } + flashGroup.lock.Unlock() + return + } + + flashGroup.lock.Unlock() + + if oldSlotStatus == proto.SlotStatus_Creating { + for _, slot := range oldPendingSlots[:updatedSlotsNum] { + c.flashNodeTopo.slotsMap[slot] = flashGroup.ID + } + } else { + c.flashNodeTopo.removeSlots(oldPendingSlots[:updatedSlotsNum]) + if needDeleteFgFlag { + c.flashNodeTopo.flashGroupMap.Delete(flashGroup.ID) + } + } + + return +} + +func (c *Cluster) checkShrinkOrDeleteFlashGroup(flashGroup *FlashGroup) (needDeleteFgFlag bool, err error) { + leftPendingSlotsNum := uint32(flashGroup.GetPendingSlotsCount()) - flashGroup.Step + if (leftPendingSlotsNum <= 0) && (flashGroup.GetPendingSlotsCount() == flashGroup.getSlotsCount()) { + needDeleteFgFlag = true + // if slots num is reduced to 0, the fn of fg need to be removed + if err = c.removeAllFlashNodeFromFlashGroup(flashGroup); err != nil { + return + } + } + return +} + +func getNewSlots(slots []uint32, pendingSlots []uint32, flag proto.SlotStatus) (newSlots []uint32) { + if flag == proto.SlotStatus_Creating { // expand flashGroup slots + newSlots = append(slots, pendingSlots...) + sort.Slice(newSlots, func(i, j int) bool { return newSlots[i] < newSlots[j] }) + return + } else { // shrink flashGroup slots + slotMap := make(map[uint32]struct{}) + for _, val := range pendingSlots { + if _, ok := slotMap[val]; !ok { + slotMap[val] = struct{}{} + } + } + for _, val := range slots { + if _, ok := slotMap[val]; !ok { + newSlots = append(newSlots, val) + } + } + return + } +} + type cTask struct { name string tickTime time.Duration @@ -502,7 +632,73 @@ func (c *Cluster) runTask(task *cTask) { } func (c *Cluster) masterAddr() (addr string) { - //TODO - //return c.leaderInfo.addr - return "tmp" + return c.leaderInfo.addr +} + +func (c *Cluster) allFlashNodes() (flashNodes []proto.NodeView) { + flashNodes = make([]proto.NodeView, 0) + c.flashNodeTopo.flashNodeMap.Range(func(addr, node interface{}) bool { + flashNode := node.(*FlashNode) + isWritable := flashNode.isWriteable() + flashNode.RLock() + flashNodes = append(flashNodes, proto.NodeView{ + ID: flashNode.ID, + Addr: flashNode.Addr, + Status: flashNode.IsActive, + IsWritable: isWritable, + }) + flashNode.RUnlock() + return true + }) + 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 (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) syncAddFlashNode(flashNode *FlashNode) (err error) { + return c.syncPutFlashNodeInfo(opSyncAddFlashNode, flashNode) +} + +func (c *Cluster) syncDeleteFlashNode(flashNode *FlashNode) (err error) { + return c.syncPutFlashNodeInfo(opSyncDeleteFlashNode, flashNode) +} + +func (c *Cluster) syncUpdateFlashNode(flashNode *FlashNode) (err error) { + return c.syncPutFlashNodeInfo(opSyncUpdateFlashNode, flashNode) +} + +func (c *Cluster) tryToChangeLeaderByHost() error { + return c.partition.TryToLeader(1) } diff --git a/remotecache/flashgroupmanager/config.go b/remotecache/flashgroupmanager/config.go index bbff3148f..9d6797677 100644 --- a/remotecache/flashgroupmanager/config.go +++ b/remotecache/flashgroupmanager/config.go @@ -1,13 +1,36 @@ package flashgroupmanager +import ( + "fmt" + syslog "log" + "strconv" + "strings" + + "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" + "github.com/cubefs/cubefs/raftstore" +) + +const ( + commaSplit = "," + colonSplit = ":" +) const ( defaultFlashNodeHandleReadTimeout = 1000 defaultFlashNodeReadDataNodeTimeout = 3000 + defaultHttpReversePoolSize = 1024 + defaultRetainLogs = 20000 ) +var AddrDatabase = make(map[uint64]string) + type clusterConfig struct { flashNodeHandleReadTimeout int flashNodeReadDataNodeTimeout int + httpProxyPoolSize uint64 + heartbeatPort int64 + replicaPort int64 + peerAddrs []string + peers []raftstore.PeerAddress } func newClusterConfig() (cfg *clusterConfig) { @@ -17,3 +40,33 @@ func newClusterConfig() (cfg *clusterConfig) { cfg.flashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout return } + +func parsePeerAddr(peerAddr string) (id uint64, ip string, port uint64, err error) { + peerStr := strings.Split(peerAddr, colonSplit) + id, err = strconv.ParseUint(peerStr[0], 10, 64) + if err != nil { + return + } + port, err = strconv.ParseUint(peerStr[2], 10, 64) + if err != nil { + return + } + ip = peerStr[1] + return +} + +func (cfg *clusterConfig) parsePeers(peerStr string) error { + peerArr := strings.Split(peerStr, commaSplit) + cfg.peerAddrs = peerArr + for _, peerAddr := range peerArr { + id, ip, port, err := parsePeerAddr(peerAddr) + if err != nil { + return err + } + cfg.peers = append(cfg.peers, raftstore.PeerAddress{Peer: proto.Peer{ID: id}, Address: ip, HeartbeatPort: int(cfg.heartbeatPort), ReplicaPort: int(cfg.replicaPort)}) + address := fmt.Sprintf("%v:%v", ip, port) + syslog.Println(address) + AddrDatabase[id] = address + } + return nil +} diff --git a/remotecache/flashgroupmanager/const.go b/remotecache/flashgroupmanager/const.go new file mode 100644 index 000000000..6bc6ca8e8 --- /dev/null +++ b/remotecache/flashgroupmanager/const.go @@ -0,0 +1,34 @@ +package flashgroupmanager + +import ( + "github.com/cubefs/cubefs/util" +) + +const ( + LRUCacheSize = 3 << 30 + WriteBufferSize = 4 * util.MB +) + +const ( + idKey = "id" + addrKey = "addr" +) + +const ( + keySeparator = "#" + clusterAcronym = "c" + maxCommonIDKey = keySeparator + "max_common_id" + clusterPrefix = keySeparator + clusterAcronym + keySeparator + flashNodePrefix = keySeparator + "fn" + keySeparator + flashGroupPrefix = keySeparator + "fg" + keySeparator +) +const ( + opSyncAllocCommonID uint32 = 0x0C + opSyncAddFlashNode uint32 = 0x6A + opSyncDeleteFlashNode uint32 = 0x6B + opSyncUpdateFlashNode uint32 = 0x6C + opSyncAddFlashGroup uint32 = 0x6D + opSyncDeleteFlashGroup uint32 = 0x6E + opSyncUpdateFlashGroup uint32 = 0x6F + opSyncPutCluster uint32 = 0x0D +) diff --git a/remotecache/flashgroupmanager/flash_group.go b/remotecache/flashgroupmanager/flash_group.go index b8604bd22..2ef15ecb9 100644 --- a/remotecache/flashgroupmanager/flash_group.go +++ b/remotecache/flashgroupmanager/flash_group.go @@ -151,3 +151,23 @@ func (fg *FlashGroup) getTargetZoneFlashNodeHosts(targetZone string) (hosts []st fg.lock.RUnlock() return } + +func NewFlashGroupFromFgv(fgv flashGroupValue) *FlashGroup { + fg := new(FlashGroup) + fg.ID = fgv.ID + fg.Slots = fgv.Slots + fg.SlotStatus = fgv.SlotStatus + fg.PendingSlots = fgv.PendingSlots + fg.Step = fgv.Step + fg.Weight = fgv.Weight + fg.Status = fgv.Status + fg.flashNodes = make(map[string]*FlashNode) + return fg +} + +func (fg *FlashGroup) GetPendingSlotsCount() (count int) { + fg.lock.RLock() + count = len(fg.PendingSlots) + fg.lock.RUnlock() + return +} diff --git a/remotecache/flashgroupmanager/flash_node_topology.go b/remotecache/flashgroupmanager/flash_node_topology.go index b87fbdfce..8dcd75374 100644 --- a/remotecache/flashgroupmanager/flash_node_topology.go +++ b/remotecache/flashgroupmanager/flash_node_topology.go @@ -99,11 +99,10 @@ func (t *FlashNodeTopology) gradualCreateFlashGroup(fgID uint64, c *Cluster, set addedSlotsNum = uint32(len(slots)) flashGroup = newFlashGroup(fgID, slots, proto.SlotStatus_Completed, make([]uint32, 0), 0, proto.FlashGroupStatus_Inactive, setWeight) } - // TODO - //if err = c.syncAddFlashGroup(flashGroup); err != nil { - // t.removeSlots(slots) - // return - //} + if err = c.syncAddFlashGroup(flashGroup); err != nil { + t.removeSlots(slots) + return + } t.flashGroupMap.Store(flashGroup.ID, flashGroup) for _, slot := range slots[:addedSlotsNum] { @@ -120,11 +119,10 @@ func (t *FlashNodeTopology) createFlashGroup(fgID uint64, c *Cluster, setSlots [ sort.Slice(slots, func(i, j int) bool { return slots[i] < slots[j] }) flashGroup = newFlashGroup(fgID, slots, proto.SlotStatus_Completed, make([]uint32, 0), 0, proto.FlashGroupStatus_Inactive, setWeight) - // TODO - //if err = c.syncAddFlashGroup(flashGroup); err != nil { - // t.removeSlots(slots) - // return - //} + if err = c.syncAddFlashGroup(flashGroup); err != nil { + t.removeSlots(slots) + return + } t.flashGroupMap.Store(flashGroup.ID, flashGroup) for _, slot := range slots { t.slotsMap[slot] = flashGroup.ID @@ -138,14 +136,13 @@ func (t *FlashNodeTopology) removeFlashGroup(flashGroup *FlashGroup, c *Cluster) flashGroup.lock.Lock() slots := flashGroup.Slots - //oldStatus := flashGroup.Status + oldStatus := flashGroup.Status flashGroup.Status = proto.FlashGroupStatus_Inactive - //TODO - //if err = c.syncDeleteFlashGroup(flashGroup); err != nil { - // flashGroup.Status = oldStatus - // flashGroup.lock.Unlock() - // return - //} + if err = c.syncDeleteFlashGroup(flashGroup); err != nil { + flashGroup.Status = oldStatus + flashGroup.lock.Unlock() + return + } flashGroup.lock.Unlock() t.removeSlots(slots) @@ -248,20 +245,19 @@ func (t *FlashNodeTopology) gradualRemoveFlashGroup(flashGroup *FlashGroup, c *C func (t *FlashNodeTopology) gradualExpandOrShrinkFlashGroupSlots(flashGroup *FlashGroup, c *Cluster, newSlotStatus proto.SlotStatus, pendingSlots []uint32, step uint32) (err error) { flashGroup.lock.Lock() - //oldSlotStatus := flashGroup.SlotStatus - //oldStep := flashGroup.Step - //oldPendingSlots := flashGroup.PendingSlots + oldSlotStatus := flashGroup.SlotStatus + oldStep := flashGroup.Step + oldPendingSlots := flashGroup.PendingSlots flashGroup.SlotStatus = newSlotStatus flashGroup.PendingSlots = pendingSlots flashGroup.Step = step - //TODO: - //if err = c.syncUpdateFlashGroup(flashGroup); err != nil { - // flashGroup.SlotStatus = oldSlotStatus - // flashGroup.PendingSlots = oldPendingSlots - // flashGroup.Step = oldStep - // flashGroup.lock.Unlock() - // return - //} + if err = c.syncUpdateFlashGroup(flashGroup); err != nil { + flashGroup.SlotStatus = oldSlotStatus + flashGroup.PendingSlots = oldPendingSlots + flashGroup.Step = oldStep + flashGroup.lock.Unlock() + return + } flashGroup.lock.Unlock() @@ -329,3 +325,21 @@ func (t *FlashNodeTopology) deleteFlashNode(flashNode *FlashNode) { } zone.flashNode.Delete(flashNode.Addr) } + +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, node interface{}) bool { + t.flashNodeMap.Delete(key) + flashNode := node.(*FlashNode) + flashNode.clean() + return true + }) + t.clientCache.Store([]byte(nil)) +} diff --git a/remotecache/flashgroupmanager/http_server.go b/remotecache/flashgroupmanager/http_server.go index 5d112d2b9..5eb6df544 100644 --- a/remotecache/flashgroupmanager/http_server.go +++ b/remotecache/flashgroupmanager/http_server.go @@ -4,6 +4,7 @@ import ( "fmt" "github.com/cubefs/cubefs/proto" "net/http" + "net/http/httputil" "time" "github.com/cubefs/cubefs/util/config" @@ -50,6 +51,19 @@ func (m *FlashGroupManager) registerAPIRoutes(router *mux.Router) { Methods(http.MethodGet). Path(proto.AdminGetIP). HandlerFunc(m.getIPAddr) + router.NewRoute().Methods(http.MethodGet, http.MethodPost). + Path(proto.AdminChangeMasterLeader). + HandlerFunc(m.changeMasterLeader) + + router.NewRoute().Methods(http.MethodGet, http.MethodPost). + Path(proto.AddRaftNode). + HandlerFunc(m.addRaftNode) + router.NewRoute().Methods(http.MethodGet, http.MethodPost). + Path(proto.RemoveRaftNode). + HandlerFunc(m.removeRaftNode) + router.NewRoute().Methods(http.MethodGet, http.MethodPost). + Path(proto.RaftStatus). + HandlerFunc(m.getRaftStatus) // APIs for FlashGroup router.NewRoute().Methods(http.MethodGet, http.MethodPost).Path(proto.AdminFlashGroupTurn).HandlerFunc(m.turnFlashGroup) @@ -141,3 +155,20 @@ func (m *FlashGroupManager) registerAPIRoutes(router *mux.Router) { // } // route.Use(interceptor) //} + +func (m *FlashGroupManager) newReverseProxy() *httputil.ReverseProxy { + tr := &http.Transport{} + if m.config != nil { + tr = proto.GetHttpTransporter(&proto.HttpCfg{ + PoolSize: int(m.config.httpProxyPoolSize), + }) + } + + return &httputil.ReverseProxy{ + Director: func(request *http.Request) { + request.URL.Scheme = "http" + request.URL.Host = m.leaderInfo.addr + }, + Transport: tr, + } +} diff --git a/remotecache/flashgroupmanager/id_allocator.go b/remotecache/flashgroupmanager/id_allocator.go index 2ab8a1878..0a9c93b49 100644 --- a/remotecache/flashgroupmanager/id_allocator.go +++ b/remotecache/flashgroupmanager/id_allocator.go @@ -1,48 +1,77 @@ package flashgroupmanager import ( + "fmt" + "strconv" "sync" "sync/atomic" + + "github.com/cubefs/cubefs/raftstore" + "github.com/cubefs/cubefs/raftstore/raftstore_db" + "github.com/cubefs/cubefs/util/log" ) type IDAllocator struct { mnIDLock sync.RWMutex commonID uint64 + + store *raftstore_db.RocksDBStore + partition raftstore.Partition } -// TODO -// func newIDAllocator(store *raftstore_db.RocksDBStore, partition raftstore.Partition) (alloc *IDAllocator) { -func newIDAllocator() (alloc *IDAllocator) { +func newIDAllocator(store *raftstore_db.RocksDBStore, partition raftstore.Partition) (alloc *IDAllocator) { alloc = new(IDAllocator) + alloc.store = store + alloc.partition = partition return } func (alloc *IDAllocator) allocateCommonID() (id uint64, err error) { alloc.mnIDLock.Lock() defer alloc.mnIDLock.Unlock() - // TDOD - //var cmd []byte - //metadata := new(RaftCmd) - //metadata.Op = opSyncAllocCommonID - //metadata.K = maxCommonIDKey + var cmd []byte + metadata := new(RaftCmd) + metadata.Op = opSyncAllocCommonID + metadata.K = maxCommonIDKey id = atomic.LoadUint64(&alloc.commonID) + 1 - //value := strconv.FormatUint(uint64(id), 10) - //metadata.V = []byte(value) - //cmd, err = metadata.Marshal() - //if err != nil { - // goto errHandler - //} - //if _, err = alloc.partition.Submit(cmd); err != nil { - // goto errHandler - //} + value := strconv.FormatUint(uint64(id), 10) + metadata.V = []byte(value) + cmd, err = metadata.Marshal() + if err != nil { + goto errHandler + } + if _, err = alloc.partition.Submit(cmd); err != nil { + goto errHandler + } alloc.setCommonID(id) return - // errHandler: - // - // log.LogErrorf("action[allocateCommonID] err:%v", err.Error()) - // return +errHandler: + log.LogErrorf("action[allocateCommonID] err:%v", err.Error()) + return } func (alloc *IDAllocator) setCommonID(id uint64) { atomic.StoreUint64(&alloc.commonID, id) } + +func (alloc *IDAllocator) restoreMaxCommonID() { + value, err := alloc.store.Get(maxCommonIDKey) + if err != nil { + panic(fmt.Sprintf("Failed to restore maxCommonID,err:%v ", err.Error())) + } + bytes := value.([]byte) + if len(bytes) == 0 { + alloc.commonID = 0 + return + } + maxMetaNodeID, err := strconv.ParseUint(string(bytes), 10, 64) + if err != nil { + panic(fmt.Sprintf("Failed to restore maxCommonID,err:%v ", err.Error())) + } + alloc.commonID = maxMetaNodeID + log.LogInfof("action[restoreMaxCommonID] maxCommonID[%v]", alloc.commonID) +} + +func (alloc *IDAllocator) restore() { + alloc.restoreMaxCommonID() +} diff --git a/remotecache/flashgroupmanager/master_manager.go b/remotecache/flashgroupmanager/master_manager.go new file mode 100644 index 000000000..96965ffa1 --- /dev/null +++ b/remotecache/flashgroupmanager/master_manager.go @@ -0,0 +1,128 @@ +package flashgroupmanager + +import ( + "fmt" + "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" + syslog "log" + "strings" + + "github.com/cubefs/cubefs/util/log" +) + +type LeaderInfo struct { + addr string //host:port + id uint64 +} + +func (m *FlashGroupManager) handleLeaderChange(leader uint64) { + if leader == 0 { + log.LogWarnf("action[handleLeaderChange] but no leader") + m.leaderInfo.id = 0 + m.leaderInfo.addr = "" + return + } + + m.leaderInfo.addr = AddrDatabase[leader] + m.leaderInfo.id = leader + + log.LogWarnf("action[handleLeaderChange] current id [%v] new leader addr [%v] leader id [%v]", m.id, m.leaderInfo.addr, leader) + m.reverseProxy = m.newReverseProxy() + + if m.id == leader { + Warn(m.clusterName, fmt.Sprintf("clusterID[%v] current is leader, leader is changed to %v", + m.clusterName, m.leaderInfo.addr)) + m.loadMetadata() + m.metaReady = true + } else { + Warn(m.clusterName, fmt.Sprintf("clusterID[%v] leader is changed to %v", + m.clusterName, m.leaderInfo.addr)) + m.clearMetadata() + m.metaReady = false + } +} + +func (m *FlashGroupManager) clearMetadata() { + m.cluster.flashNodeTopo.clear() + m.cluster.flashNodeTopo = NewFlashNodeTopology() +} + +func (m *FlashGroupManager) loadMetadata() { + var err error + log.LogInfo("action[loadMetadata] begin") + syslog.Println("action[loadMetadata] begin") + m.clearMetadata() + m.restoreIDAlloc() + m.cluster.fsm.restore() + + if err = m.cluster.loadClusterValue(); err != nil { + panic(err) + } + + if err = m.cluster.loadFlashNodes(); err != nil { + panic(err) + } + + if err = m.cluster.loadFlashGroups(); err != nil { + panic(err) + } + + if err = m.cluster.loadFlashTopology(); err != nil { + panic(err) + } + + // TODO + //log.LogInfo("action[loadApiLimiterInfo] begin") + //if err = m.cluster.loadApiLimiterInfo(); err != nil { + // panic(err) + //} + //log.LogInfo("action[loadApiLimiterInfo] end") + + log.LogInfo("action[loadMetadata] end") + syslog.Println("action[loadMetadata] end") +} + +func (m *FlashGroupManager) restoreIDAlloc() { + m.cluster.idAlloc.restore() +} + +func (m *FlashGroupManager) handlePeerChange(confChange *proto.ConfChange) (err error) { + var msg string + addr := string(confChange.Context) + switch confChange.Type { + case proto.ConfAddNode: + var arr []string + if arr = strings.Split(addr, colonSplit); len(arr) < 2 { + msg = fmt.Sprintf("action[handlePeerChange] clusterID[%v] nodeAddr[%v] is invalid", m.clusterName, addr) + break + } + m.raftStore.AddNodeWithPort(confChange.Peer.ID, arr[0], int(m.config.heartbeatPort), int(m.config.replicaPort)) + AddrDatabase[confChange.Peer.ID] = string(confChange.Context) + msg = fmt.Sprintf("clusterID[%v] peerID:%v,nodeAddr[%v] has been add", m.clusterName, confChange.Peer.ID, addr) + case proto.ConfRemoveNode: + m.raftStore.DeleteNode(confChange.Peer.ID) + msg = fmt.Sprintf("clusterID[%v] peerID:%v,nodeAddr[%v] has been removed", m.clusterName, confChange.Peer.ID, addr) + default: + // do nothing + } + Warn(m.clusterName, msg) + return +} + +func (m *FlashGroupManager) handleApplySnapshot() { + m.fsm.restore() + m.restoreIDAlloc() +} + +func (m *FlashGroupManager) handleRaftUserCmd(opt uint32, key string, cmdMap map[string][]byte) (err error) { + log.LogInfof("action[handleRaftUserCmd] opt %v, key %v, map len %v", opt, key, len(cmdMap)) + switch opt { + // TODO + //case opSyncPutFollowerApiLimiterInfo, opSyncPutApiLimiterInfo: + // if m.cluster != nil && !m.partition.IsRaftLeader() { + // m.cluster.apiLimiter.updateLimiterInfoFromLeader(cmdMap[key]) + // } + default: + log.LogErrorf("action[handleRaftUserCmd] opt %v not supported,key %v, map len %v", opt, key, len(cmdMap)) + } + return nil +} diff --git a/remotecache/flashgroupmanager/metadata_fsm.go b/remotecache/flashgroupmanager/metadata_fsm.go new file mode 100644 index 000000000..394500a71 --- /dev/null +++ b/remotecache/flashgroupmanager/metadata_fsm.go @@ -0,0 +1,302 @@ +package flashgroupmanager + +import ( + "encoding/json" + "fmt" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/stat" + "io" + "strconv" + "sync" + + "github.com/cubefs/cubefs/depends/tiglabs/raft" + "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" + raftProto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto" + raftstore "github.com/cubefs/cubefs/raftstore/raftstore_db" + "github.com/cubefs/cubefs/util/errors" + "github.com/cubefs/cubefs/util/log" +) + +const ( + applied = "applied" +) + +type raftLeaderChangeHandler func(leader uint64) + +type raftPeerChangeHandler func(confChange *proto.ConfChange) (err error) + +type raftUserCmdApplyHandler func(opt uint32, key string, cmdMap map[string][]byte) (err error) + +type raftApplySnapshotHandler func() + +type MetadataFsm struct { + store *raftstore.RocksDBStore + rs *raft.RaftServer + applied uint64 + retainLogs uint64 + leaderChangeHandler raftLeaderChangeHandler + peerChangeHandler raftPeerChangeHandler + snapshotHandler raftApplySnapshotHandler + UserAppCmdHandler raftUserCmdApplyHandler + onSnapshot bool + raftLk sync.Mutex +} + +func newMetadataFsm(store *raftstore.RocksDBStore, retainsLog uint64, rs *raft.RaftServer) (fsm *MetadataFsm) { + fsm = new(MetadataFsm) + fsm.store = store + fsm.rs = rs + fsm.retainLogs = retainsLog + return +} + +// Corresponding to the LeaderChange interface in Raft library. +func (mf *MetadataFsm) registerLeaderChangeHandler(handler raftLeaderChangeHandler) { + mf.leaderChangeHandler = handler +} + +// Corresponding to the PeerChange interface in Raft library. +func (mf *MetadataFsm) registerPeerChangeHandler(handler raftPeerChangeHandler) { + mf.peerChangeHandler = handler +} + +// Corresponding to the ApplySnapshot interface in Raft library. +func (mf *MetadataFsm) registerApplySnapshotHandler(handler raftApplySnapshotHandler) { + mf.snapshotHandler = handler +} + +// Corresponding to the ApplyRaftCmd interface in Raft library. +func (mf *MetadataFsm) registerRaftUserCmdApplyHandler(handler raftUserCmdApplyHandler) { + mf.UserAppCmdHandler = handler +} + +func (mf *MetadataFsm) restore() { + mf.restoreApplied() +} + +func (mf *MetadataFsm) restoreApplied() { + value, err := mf.store.Get(applied) + if err != nil { + panic(fmt.Sprintf("Failed to restore applied err:%v", err.Error())) + } + byteValues := value.([]byte) + if len(byteValues) == 0 { + mf.applied = 0 + return + } + applied, err := strconv.ParseUint(string(byteValues), 10, 64) + if err != nil { + panic(fmt.Sprintf("Failed to restore applied,err:%v ", err.Error())) + } + mf.applied = applied + log.LogInfof("[restoreApplied] apply index(%v)", applied) +} + +func (c *Cluster) addRaftNode(nodeID uint64, addr string) (err error) { + log.LogInfof("action[addRaftNode] nodeID: %v, addr: %v:", nodeID, addr) + + peer := raftProto.Peer{ID: nodeID} + _, err = c.partition.ChangeMember(raftProto.ConfAddNode, peer, []byte(addr)) + if err != nil { + return errors.New("action[addRaftNode] error: " + err.Error()) + } + return nil +} + +func (c *Cluster) removeRaftNode(nodeID uint64, addr string) (err error) { + log.LogInfof("action[removeRaftNode] nodeID: %v, addr: %v:", nodeID, addr) + + peer := raftProto.Peer{ID: nodeID} + _, err = c.partition.ChangeMember(raftProto.ConfRemoveNode, peer, []byte(addr)) + if err != nil { + return errors.New("action[removeRaftNode] error: " + err.Error()) + } + return nil +} + +// Apply implements the interface of raft.StateMachine +func (mf *MetadataFsm) Apply(command []byte, index uint64) (resp interface{}, err error) { + mf.raftLk.Lock() + defer mf.raftLk.Unlock() + + cmd := new(RaftCmd) + if err = cmd.Unmarshal(command); err != nil { + log.LogErrorf("action[fsmApply],unmarshal data:%v, err:%v", command, err.Error()) + panic(err) + } + + log.LogDebugf("action[Apply] apply index(%v), op %d", index, cmd.Op) + + cmdMap := make(map[string][]byte) + deleteSet := make(map[string]util.Null) + cmdMap[cmd.K] = cmd.V + cmdMap[applied] = []byte(strconv.FormatUint(uint64(index), 10)) + + switch cmd.Op { + case opSyncDeleteFlashNode, opSyncDeleteFlashGroup: + if err = mf.delKeyAndPutIndex(cmd.K, cmdMap); err != nil { + panic(err) + } + // TODO + //case opSyncPutFollowerApiLimiterInfo, opSyncPutApiLimiterInfo: + // mf.UserAppCmdHandler(cmd.Op, cmd.K, cmdMap) + // //if err = mf.delKeyAndPutIndex(cmd.K, cmdMap); err != nil { + // // panic(err)api_args_parse.go + // //} + // if err = mf.store.BatchPut(cmdMap, true); err != nil { + // panic(err) + // } + default: + // sync put data + if err = mf.store.BatchDeleteAndPut(deleteSet, cmdMap, true); err != nil { + panic(err) + } + } + log.LogDebugf("action[Apply],persist index[%v]", string(cmdMap[applied])) + mf.applied = index + + if mf.applied > 0 && (mf.applied%mf.retainLogs) == 0 { + log.LogWarnf("action[Apply],truncate raft log,retainLogs[%v],index[%v]", mf.retainLogs, mf.applied) + mf.rs.Truncate(GroupID, mf.applied) + } + return +} + +// ApplyMemberChange implements the interface of raft.StateMachine +func (mf *MetadataFsm) ApplyMemberChange(confChange *proto.ConfChange, index uint64) (interface{}, error) { + mf.raftLk.Lock() + defer mf.raftLk.Unlock() + + var err error + if mf.peerChangeHandler != nil { + err = mf.peerChangeHandler(confChange) + } + return nil, err +} + +// Snapshot implements the interface of raft.StateMachine +func (mf *MetadataFsm) Snapshot() (proto.Snapshot, error) { + mf.raftLk.Lock() + defer mf.raftLk.Unlock() + + snapshot := mf.store.RocksDBSnapshot() + iterator := mf.store.Iterator(snapshot) + iterator.SeekToFirst() + return &MetadataSnapshot{ + applied: mf.applied, + snapshot: snapshot, + fsm: mf, + iterator: iterator, + }, nil +} + +func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIterator) (err error) { + log.LogWarnf("action[ApplySnapshot] reset rocksdb before applying snapshot") + mf.onSnapshot = true + + defer func() { + mf.onSnapshot = false + }() + + if log.EnableDebug() { + func() { + snap := mf.store.RocksDBSnapshot() + defer mf.store.ReleaseSnapshot(snap) + iter := mf.store.Iterator(snap) + defer iter.Close() + cnt := 0 + for iter.SeekToFirst(); iter.Valid(); iter.Next() { + cnt++ + } + log.LogDebugf("[ApplySnapshot] scan %v keys before clear", cnt) + }() + } + + mf.store.Clear() + + if log.EnableDebug() { + func() { + snap := mf.store.RocksDBSnapshot() + defer mf.store.ReleaseSnapshot(snap) + iter := mf.store.Iterator(snap) + defer iter.Close() + cnt := 0 + for iter.SeekToFirst(); iter.Valid(); iter.Next() { + cnt++ + } + log.LogDebugf("[ApplySnapshot] scan %v keys after clear", cnt) + }() + } + + log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] begin,applied[%v]", mf.applied)) + var data []byte + var appliedIndex []byte + for err == nil { + bgTime := stat.BeginStat() + if data, err = iterator.Next(); err != nil { + break + } + stat.EndStat("ApplySnapshot-Next", err, bgTime, 1) + cmd := &RaftCmd{} + if err = json.Unmarshal(data, cmd); err != nil { + goto errHandler + } + bgTime = stat.BeginStat() + if cmd.K != applied { + if _, err = mf.store.Put(cmd.K, cmd.V, false); err != nil { + goto errHandler + } + } else { + appliedIndex = cmd.V + } + stat.EndStat("ApplySnapshot-Put", err, bgTime, 1) + } + if err != nil && err != io.EOF { + goto errHandler + } + + if err = mf.store.Flush(); err != nil { + log.LogError(fmt.Sprintf("action[ApplySnapshot] Flush failed,err:%v", err.Error())) + goto errHandler + } + + // NOTE: we write applied index at last + log.LogDebugf("[ApplySnapshot] find applied index(%v)", appliedIndex) + if appliedIndex != nil { + if _, err = mf.store.Put(applied, appliedIndex, true); err != nil { + goto errHandler + } + } else { + log.LogErrorf("[ApplySnapshot] not found applied index in snapshot") + } + + mf.snapshotHandler() + log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] success,applied[%v]", mf.applied)) + return nil +errHandler: + log.LogError(fmt.Sprintf("action[ApplySnapshot] failed,err:%v", err.Error())) + return err +} + +// HandleFatalEvent implements the interface of raft.StateMachine +func (mf *MetadataFsm) HandleFatalEvent(err *raft.FatalError) { + panic(err.Err) +} + +// HandleLeaderChange implements the interface of raft.StateMachine +func (mf *MetadataFsm) HandleLeaderChange(leader uint64) { + if mf.leaderChangeHandler != nil { + go mf.leaderChangeHandler(leader) + } +} + +func (mf *MetadataFsm) delKeyAndPutIndex(key string, cmdMap map[string][]byte) (err error) { + return mf.store.DeleteKeyAndPutIndex(key, cmdMap, true) +} + +// Stop stops the RaftServer +func (mf *MetadataFsm) Stop() { + if mf.rs != nil { + mf.rs.Stop() + } +} diff --git a/remotecache/flashgroupmanager/metadata_fsm_op.go b/remotecache/flashgroupmanager/metadata_fsm_op.go new file mode 100644 index 000000000..5ab27c072 --- /dev/null +++ b/remotecache/flashgroupmanager/metadata_fsm_op.go @@ -0,0 +1,175 @@ +package flashgroupmanager + +import ( + "encoding/json" + "fmt" + "strings" + + "github.com/cubefs/cubefs/util/errors" + "github.com/cubefs/cubefs/util/log" +) + +type RaftCmd struct { + Op uint32 `json:"op"` + K string `json:"k"` + V []byte `json:"v"` +} + +func (m *RaftCmd) Marshal() ([]byte, error) { + return json.Marshal(m) +} + +// Unmarshal converts the byte array to a RaftCmd. +func (m *RaftCmd) Unmarshal(data []byte) (err error) { + return json.Unmarshal(data, m) +} + +type clusterValue struct { + Name string + FlashNodeHandleReadTimeout int + FlashNodeReadDataNodeTimeout int +} + +func newClusterValue(c *Cluster) (cv *clusterValue) { + cv = &clusterValue{ + Name: c.Name, + FlashNodeHandleReadTimeout: c.cfg.flashNodeHandleReadTimeout, + FlashNodeReadDataNodeTimeout: c.cfg.flashNodeReadDataNodeTimeout, + } + return cv +} + +func (c *Cluster) loadClusterValue() (err error) { + result, err := c.fsm.store.SeekForPrefix([]byte(clusterPrefix)) + if err != nil { + err = fmt.Errorf("action[loadClusterValue],err:%v", err.Error()) + return err + } + for _, value := range result { + cv := &clusterValue{} + if err = json.Unmarshal(value, cv); err != nil { + log.LogErrorf("action[loadClusterValue], unmarshal err:%v", err.Error()) + return err + } + + if cv.Name != c.Name { + log.LogErrorf("action[loadClusterValue] clusterName(%v) not match loaded clusterName(%v), n loaded cluster value: %+v", + c.Name, cv.Name, cv) + continue + } + + log.LogDebugf("action[loadClusterValue] loaded cluster value: %+v", cv) + + if cv.FlashNodeHandleReadTimeout == 0 { + cv.FlashNodeHandleReadTimeout = defaultFlashNodeHandleReadTimeout + } + c.cfg.flashNodeHandleReadTimeout = cv.FlashNodeHandleReadTimeout + + if cv.FlashNodeReadDataNodeTimeout == 0 { + cv.FlashNodeReadDataNodeTimeout = defaultFlashNodeReadDataNodeTimeout + } + c.cfg.flashNodeReadDataNodeTimeout = cv.FlashNodeReadDataNodeTimeout + log.LogInfof("action[loadClusterValue] flashNodeHandleReadTimeout %v(ms), flashNodeReadDataNodeTimeout%v(ms)", + cv.FlashNodeHandleReadTimeout, cv.FlashNodeReadDataNodeTimeout) + } + + 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 + // load later in loadFlashTopology + flashNode.FlashGroupID = fnv.FlashGroupID + + _, err = c.flashNodeTopo.getZone(flashNode.ZoneName) + if err != nil { + c.flashNodeTopo.putZoneIfAbsent(NewFlashNodeZone(flashNode.ZoneName)) + err = nil + } + c.flashNodeTopo.putFlashNode(flashNode) + log.LogInfof("action[loadFlashNodes], flashNode[flashNodeId:%v addr:%s flashGroupId:%v]", flashNode.ID, 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 := NewFlashGroupFromFgv(fgv) + 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 +} + +func (c *Cluster) loadFlashTopology() (err error) { + c.flashNodeTopo.flashNodeMap.Range(func(addr, flashNode interface{}) bool { + node := flashNode.(*FlashNode) + node.Lock() + if gid := node.FlashGroupID; gid != UnusedFlashNodeFlashGroupID { + if g, e := c.flashNodeTopo.getFlashGroup(gid); e == nil { + g.putFlashNode(node) + log.LogInfof("action[loadFlashTopology] load FlashNode[%s] -> FlashGroup[%d]", node.Addr, gid) + } else { + node.FlashGroupID = UnusedFlashNodeFlashGroupID + log.LogErrorf("action[loadFlashTopology] FlashNode[flashNodeId:%v addr:%s flashGroupId:%v] err:%v", node.ID, node.Addr, node.FlashGroupID, e.Error()) + } + } + node.Unlock() + return true + }) + return +} + +func (m *RaftCmd) setOpType() { + keyArr := strings.Split(m.K, keySeparator) + if len(keyArr) < 2 { + log.LogWarnf("action[setOpType] invalid length[%v]", keyArr) + return + } + switch keyArr[1] { + case clusterAcronym: + m.Op = opSyncPutCluster + case maxCommonIDKey: + m.Op = opSyncAllocCommonID + default: + log.LogWarnf("action[setOpType] unknown opCode[%v]", keyArr[1]) + } +} + +func (c *Cluster) submit(metadata *RaftCmd) (err error) { + cmd, err := metadata.Marshal() + if err != nil { + return errors.New(err.Error()) + } + if _, err = c.partition.Submit(cmd); err != nil { + msg := fmt.Sprintf("action[metadata_submit] err:%v", err.Error()) + return errors.New(msg) + } + return +} diff --git a/remotecache/flashgroupmanager/metadata_snapshot.go b/remotecache/flashgroupmanager/metadata_snapshot.go new file mode 100644 index 000000000..cd1a28869 --- /dev/null +++ b/remotecache/flashgroupmanager/metadata_snapshot.go @@ -0,0 +1,47 @@ +package flashgroupmanager + +import ( + "fmt" + "io" + + "github.com/tecbot/gorocksdb" +) + +// MetadataSnapshot represents the snapshot of a meta partition +type MetadataSnapshot struct { + fsm *MetadataFsm + applied uint64 + snapshot *gorocksdb.Snapshot + iterator *gorocksdb.Iterator +} + +// ApplyIndex implements the Snapshot interface +func (ms *MetadataSnapshot) ApplyIndex() uint64 { + return ms.applied +} + +// Close implements the Snapshot interface +func (ms *MetadataSnapshot) Close() { + ms.fsm.store.ReleaseSnapshot(ms.snapshot) +} + +// Next implements the Snapshot interface +func (ms *MetadataSnapshot) Next() (data []byte, err error) { + md := new(RaftCmd) + if ms.iterator.Valid() { + key := ms.iterator.Key() + md.K = string(key.Data()) + md.setOpType() + value := ms.iterator.Value() + if value != nil { + md.V = value.Data() + } + if data, err = md.Marshal(); err != nil { + err = fmt.Errorf("action[Next],marshal kv:%v,err:%v", md, err.Error()) + return nil, err + } + ms.iterator.Next() + return data, nil + } + return nil, io.EOF +} diff --git a/remotecache/flashgroupmanager/monitor_metrics.go b/remotecache/flashgroupmanager/monitor_metrics.go new file mode 100644 index 000000000..70f12c910 --- /dev/null +++ b/remotecache/flashgroupmanager/monitor_metrics.go @@ -0,0 +1 @@ +package flashgroupmanager diff --git a/remotecache/flashgroupmanager/operate_util.go b/remotecache/flashgroupmanager/operate_util.go index 8398c1542..f079f0667 100644 --- a/remotecache/flashgroupmanager/operate_util.go +++ b/remotecache/flashgroupmanager/operate_util.go @@ -29,6 +29,10 @@ func unmatchedKey(name string) (err error) { return errors.NewErrorf("parameter %v not match", name) } +func keyNotFound(name string) (err error) { + return errors.NewErrorf("parameter %v not found", name) +} + func contains(arr []string, element string) (ok bool) { if len(arr) == 0 { return diff --git a/remotecache/flashgroupmanager/server.go b/remotecache/flashgroupmanager/server.go index 2ee432c00..ee6a92e4a 100644 --- a/remotecache/flashgroupmanager/server.go +++ b/remotecache/flashgroupmanager/server.go @@ -3,12 +3,17 @@ package flashgroupmanager import ( "context" "fmt" + "github.com/cubefs/cubefs/raftstore" + syslog "log" "net/http" + "net/http/httputil" + "strconv" "sync" "sync/atomic" "time" "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/raftstore/raftstore_db" "github.com/cubefs/cubefs/util/config" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/exporter" @@ -21,10 +26,22 @@ const ( ) const ( - ClusterName = "clusterName" - IP = "ip" - Stat = "stat" - LogDir = "logDir" + ClusterName = "clusterName" + IP = "ip" + Stat = "stat" + LogDir = "logDir" + StoreDir = "storeDir" + HttpReversePoolSize = "httpReversePoolSize" + ID = "id" + WalDir = "walDir" + RetainLogs = "retainLogs" + HeartbeatPort = "heartbeatPort" + ReplicaPort = "replicaPort" + Peers = "peers" + TickInterval = "tickInterval" + RaftRecvBufSize = "raftRecvBufSize" + ElectionTick = "electionTick" + GroupID = 1 ) const ( @@ -32,16 +49,29 @@ const ( ) type FlashGroupManager struct { - wg sync.WaitGroup - clusterName string - config *clusterConfig - ip string - bindIp bool - port string - apiServer *http.Server - cluster *Cluster - logDir string - metaReady bool //TODO leader change + wg sync.WaitGroup + clusterName string + config *clusterConfig + ip string + bindIp bool + port string + apiServer *http.Server + cluster *Cluster + logDir string + metaReady bool + leaderInfo *LeaderInfo + rocksDBStore *raftstore_db.RocksDBStore + storeDir string + reverseProxy *httputil.ReverseProxy + id uint64 + walDir string + retainLogs uint64 + tickInterval int + raftRecvBufSize int + electionTick int + raftStore raftstore.RaftStore + fsm *MetadataFsm + partition raftstore.Partition } func NewFlashGroupManager() *FlashGroupManager { @@ -55,13 +85,17 @@ func (m *FlashGroupManager) Start(cfg *config.Config) (err error) { return } - // TODO - // m.leaderInfo = &LeaderInfo{} + m.leaderInfo = &LeaderInfo{} - // TODO - // if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil { - // TODO - // if err = m.createRaftServer(cfg); err != nil { + if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil { + return + } + + m.reverseProxy = m.newReverseProxy() + if err = m.createRaftServer(cfg); err != nil { + log.LogError(errors.Stack(err)) + return + } m.initCluster() @@ -102,17 +136,15 @@ func (m *FlashGroupManager) Shutdown() { } stat.CloseStat() - // TODO - //if m.fsm != nil { - // m.fsm.Stop() - //} + if m.fsm != nil { + m.fsm.Stop() + } // NOTE: wait 10 second for background goroutines to exit time.Sleep(10 * time.Second) - // TODO - //if m.rocksDBStore != nil { - // m.rocksDBStore.Close() - //} + if m.rocksDBStore != nil { + m.rocksDBStore.Close() + } m.wg.Done() } @@ -127,22 +159,112 @@ func (m *FlashGroupManager) checkConfig(cfg *config.Config) (err error) { m.bindIp = cfg.GetBool(proto.BindIpKey) m.port = cfg.GetString(proto.ListenPort) m.logDir = cfg.GetString(LogDir) + m.walDir = cfg.GetString(WalDir) + peerAddrs := cfg.GetString(Peers) - if m.port == "" || m.clusterName == "" { - return fmt.Errorf("%v,err:%v,%v,%v", proto.ErrInvalidCfg, "one of (listen,walDir,clusterName) is null", - m.ip, m.clusterName) + 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.ip, m.clusterName, m.walDir, peerAddrs) } - return + + m.config.heartbeatPort = cfg.GetInt64(HeartbeatPort) + m.config.replicaPort = cfg.GetInt64(ReplicaPort) + if m.config.heartbeatPort <= 1024 { + m.config.heartbeatPort = raftstore.DefaultHeartbeatPort + } + if m.config.replicaPort <= 1024 { + m.config.replicaPort = raftstore.DefaultReplicaPort + } + syslog.Printf("heartbeatPort[%v],replicaPort[%v]\n", m.config.heartbeatPort, m.config.replicaPort) + if err = m.config.parsePeers(peerAddrs); err != nil { + return + } + + m.storeDir = cfg.GetString(StoreDir) + if m.storeDir == "" { + return fmt.Errorf("store dir is empty") + } + + retainLogs := cfg.GetString(RetainLogs) + if retainLogs != "" { + if m.retainLogs, err = strconv.ParseUint(retainLogs, 10, 64); err != nil { + return fmt.Errorf("%v,err:%v", proto.ErrInvalidCfg, err.Error()) + } + } + if m.retainLogs <= 0 { + m.retainLogs = defaultRetainLogs + } + syslog.Println("retainLogs=", m.retainLogs) + + if m.id, err = strconv.ParseUint(cfg.GetString(ID), 10, 64); err != nil { + return fmt.Errorf("%v,err:%v", proto.ErrInvalidCfg, err.Error()) + } + + m.config.httpProxyPoolSize = uint64(cfg.GetInt64(HttpReversePoolSize)) + if m.config.httpProxyPoolSize < defaultHttpReversePoolSize { + m.config.httpProxyPoolSize = defaultHttpReversePoolSize + } + syslog.Printf("http reverse pool size %d \n", m.config.httpProxyPoolSize) + + m.tickInterval = int(cfg.GetFloat(TickInterval)) + m.raftRecvBufSize = int(cfg.GetInt(RaftRecvBufSize)) + m.electionTick = int(cfg.GetFloat(ElectionTick)) + if m.tickInterval <= 300 { + m.tickInterval = 500 + } + if m.electionTick <= 3 { + m.electionTick = 5 + } + return nil } func (m *FlashGroupManager) initCluster() { log.LogInfo("action[initCluster] begin") - m.cluster = newCluster(m.clusterName, m.config) + m.cluster = newCluster(m.clusterName, m.config, m.leaderInfo, m.fsm, m.partition) log.LogInfo("action[initCluster] end") - // in case any limiter on follower log.LogInfo("action[loadApiLimiterInfo] begin") // TODO //m.cluster.loadApiLimiterInfo() log.LogInfo("action[loadApiLimiterInfo] end") } + +func (m *FlashGroupManager) createRaftServer(cfg *config.Config) (err error) { + raftCfg := &raftstore.Config{ + NodeID: m.id, + RaftPath: m.walDir, + IPAddr: cfg.GetString(IP), + NumOfLogsToRetain: m.retainLogs, + HeartbeatPort: int(m.config.heartbeatPort), + ReplicaPort: int(m.config.replicaPort), + TickInterval: m.tickInterval, + ElectionTick: m.electionTick, + RecvBufSize: m.raftRecvBufSize, + } + if m.raftStore, err = raftstore.NewRaftStore(raftCfg, cfg); err != nil { + return errors.Trace(err, "NewRaftStore failed! id[%v] walPath[%v]", m.id, m.walDir) + } + syslog.Printf("peers[%v],tickInterval[%v],electionTick[%v]\n", m.config.peers, m.tickInterval, m.electionTick) + m.initFsm() + partitionCfg := &raftstore.PartitionConfig{ + ID: GroupID, + Peers: m.config.peers, + Applied: m.fsm.applied, + SM: m.fsm, + } + if m.partition, err = m.raftStore.CreatePartition(partitionCfg); err != nil { + return errors.Trace(err, "CreatePartition failed") + } + return +} + +func (m *FlashGroupManager) initFsm() { + m.fsm = newMetadataFsm(m.rocksDBStore, m.retainLogs, m.raftStore.RaftServer()) + m.fsm.registerLeaderChangeHandler(m.handleLeaderChange) + m.fsm.registerPeerChangeHandler(m.handlePeerChange) + + // register the handlers for the interfaces defined in the Raft library + m.fsm.registerApplySnapshotHandler(m.handleApplySnapshot) + m.fsm.registerRaftUserCmdApplyHandler(m.handleRaftUserCmd) + m.fsm.restore() +}