mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
enhance(lcnode): support multi task
1. add lcTaskCountLimit 2. delete Scanning and add WorkingCount 3. add lcnode addr in heartbeat resp and scan resp Signed-off-by: zhaochenyang <zhaochenyang@oppo.com>
This commit is contained in:
zhaochenyang
chenyangzhao
parent
1f855b084f
commit
bd6608b8d0
@ -30,6 +30,7 @@ const (
|
||||
configLcScanLimitPerSecondStr = "lcScanLimitPerSecond"
|
||||
|
||||
configSnapshotRoutineNumPerTaskStr = "snapshotRoutineNumPerTask"
|
||||
configLcNodeTaskCountLimit = "lcNodeTaskCountLimit"
|
||||
)
|
||||
|
||||
// Default of configuration value
|
||||
@ -53,6 +54,7 @@ const (
|
||||
defaultIntervalToCheckRegister = 2 * defaultLcNodeTimeOutSec
|
||||
|
||||
defaultUnboundedChanInitCapacity = 10000
|
||||
defaultLcNodeTaskCountLimit = 1
|
||||
)
|
||||
|
||||
var (
|
||||
@ -65,4 +67,5 @@ var (
|
||||
lcScanLimitPerSecond rate.Limit
|
||||
|
||||
snapshotRoutineNumPerTask int
|
||||
lcNodeTaskCountLimit int
|
||||
)
|
||||
|
||||
@ -59,7 +59,8 @@ func (l *LcNode) opMasterHeartbeat(conn net.Conn, p *proto.Packet, remoteAddr st
|
||||
l.scannerMutex.RLock()
|
||||
for _, scanner := range l.lcScanners {
|
||||
result := &proto.LcNodeRuleTaskResponse{
|
||||
ID: scanner.ID,
|
||||
ID: scanner.ID,
|
||||
LcNode: l.localServerAddr,
|
||||
LcNodeRuleTaskStatistics: proto.LcNodeRuleTaskStatistics{
|
||||
Volume: scanner.Volume,
|
||||
RuleId: scanner.rule.ID,
|
||||
@ -74,7 +75,9 @@ func (l *LcNode) opMasterHeartbeat(conn net.Conn, p *proto.Packet, remoteAddr st
|
||||
}
|
||||
for _, scanner := range l.snapshotScanners {
|
||||
info := &proto.SnapshotVerDelTaskResponse{
|
||||
ID: scanner.ID,
|
||||
ID: scanner.ID,
|
||||
LcNode: l.localServerAddr,
|
||||
SnapshotVerDelTask: scanner.verDelReq.Task,
|
||||
SnapshotStatistics: proto.SnapshotStatistics{
|
||||
VolName: scanner.Volume,
|
||||
VerSeq: scanner.getTaskVerSeq(),
|
||||
@ -88,6 +91,7 @@ func (l *LcNode) opMasterHeartbeat(conn net.Conn, p *proto.Packet, remoteAddr st
|
||||
}
|
||||
l.scannerMutex.RUnlock()
|
||||
|
||||
resp.LcTaskCountLimit = lcNodeTaskCountLimit
|
||||
resp.Status = proto.TaskSucceeds
|
||||
|
||||
end:
|
||||
|
||||
@ -128,11 +128,10 @@ func (s *LcScanner) Start() (err error) {
|
||||
if err != nil {
|
||||
log.LogErrorf("startScan err(%v): volume(%v), rule id(%v), scanning done!",
|
||||
err, s.Volume, s.rule.ID)
|
||||
t := time.Now()
|
||||
response.ID = s.ID
|
||||
response.EndTime = &t
|
||||
response.Status = proto.TaskSucceeds
|
||||
response.Done = true
|
||||
response.LcNode = s.lcnode.localServerAddr
|
||||
response.Status = proto.TaskFailed
|
||||
response.Result = err.Error()
|
||||
|
||||
s.lcnode.scannerMutex.Lock()
|
||||
delete(s.lcnode.lcScanners, s.ID)
|
||||
@ -522,6 +521,7 @@ func (s *LcScanner) checkScanning() {
|
||||
response.Status = proto.TaskSucceeds
|
||||
response.Done = true
|
||||
response.ID = s.ID
|
||||
response.LcNode = s.lcnode.localServerAddr
|
||||
response.Volume = s.Volume
|
||||
response.RuleId = s.rule.ID
|
||||
response.ExpiredNum = s.currentStat.ExpiredNum
|
||||
|
||||
@ -195,6 +195,21 @@ func (l *LcNode) parseConfig(cfg *config.Config) (err error) {
|
||||
}
|
||||
log.LogInfof("loadConfig: setup config: %v(%v)", configLcScanLimitPerSecondStr, lcScanLimitPerSecond)
|
||||
|
||||
// parse lcNodeTaskCount
|
||||
var count int64
|
||||
countStr := cfg.GetString(configLcNodeTaskCountLimit)
|
||||
if countStr != "" {
|
||||
if count, err = strconv.ParseInt(countStr, 10, 64); err != nil {
|
||||
return fmt.Errorf("%v,err:%v", proto.ErrInvalidCfg, err.Error())
|
||||
}
|
||||
}
|
||||
if count <= 0 {
|
||||
lcNodeTaskCountLimit = defaultLcNodeTaskCountLimit
|
||||
} else {
|
||||
lcNodeTaskCountLimit = int(count)
|
||||
}
|
||||
log.LogInfof("loadConfig: setup config: %v(%v)", configLcNodeTaskCountLimit, lcNodeTaskCountLimit)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@ -418,6 +418,8 @@ func (s *SnapshotScanner) checkScanning(report bool) {
|
||||
response.Status = proto.TaskSucceeds
|
||||
response.Done = true
|
||||
response.ID = s.ID
|
||||
response.LcNode = s.lcnode.localServerAddr
|
||||
response.SnapshotVerDelTask = s.verDelReq.Task
|
||||
response.VolName = s.Volume
|
||||
response.VerSeq = s.getTaskVerSeq()
|
||||
response.FileNum = s.currentStat.FileNum
|
||||
|
||||
@ -4297,13 +4297,13 @@ func (c *Cluster) addLcNode(nodeAddr string) (id uint64, err error) {
|
||||
goto errHandler
|
||||
}
|
||||
c.lcNodes.Store(nodeAddr, ln)
|
||||
|
||||
c.lcMgr.lcNodeStatus.Lock()
|
||||
delete(c.lcMgr.lcNodeStatus.WorkingNodes, nodeAddr)
|
||||
c.lcMgr.lcNodeStatus.IdleNodes[nodeAddr] = nodeAddr
|
||||
c.lcMgr.lcNodeStatus.WorkingCount[nodeAddr] = 0
|
||||
c.lcMgr.lcNodeStatus.Unlock()
|
||||
|
||||
c.snapshotMgr.lcNodeStatus.Lock()
|
||||
delete(c.snapshotMgr.lcNodeStatus.WorkingNodes, nodeAddr)
|
||||
c.snapshotMgr.lcNodeStatus.IdleNodes[nodeAddr] = nodeAddr
|
||||
c.snapshotMgr.lcNodeStatus.WorkingCount[nodeAddr] = 0
|
||||
c.snapshotMgr.lcNodeStatus.Unlock()
|
||||
log.LogInfof("action[addLcNode], clusterID[%v], lcNodeAddr: %v, id: %v, add idleNodes", c.Name, nodeAddr, ln.ID)
|
||||
return ln.ID, nil
|
||||
@ -4356,8 +4356,8 @@ func (c *Cluster) clearLcNodes() {
|
||||
}
|
||||
|
||||
func (c *Cluster) delLcNode(nodeAddr string) (err error) {
|
||||
c.lcMgr.lcRuleTaskStatus.RedoTask(c.lcMgr.lcNodeStatus.RemoveNode(nodeAddr))
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.RedoTask(c.snapshotMgr.lcNodeStatus.RemoveNode(nodeAddr))
|
||||
c.lcMgr.lcNodeStatus.RemoveNode(nodeAddr)
|
||||
c.snapshotMgr.lcNodeStatus.RemoveNode(nodeAddr)
|
||||
|
||||
lcNode, err := c.lcNode(nodeAddr)
|
||||
if err != nil {
|
||||
|
||||
@ -15,6 +15,7 @@
|
||||
package master
|
||||
|
||||
import (
|
||||
"math"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@ -36,8 +37,8 @@ func newLifecycleManager() *lifecycleManager {
|
||||
log.LogInfof("action[newLifecycleManager] construct")
|
||||
lcMgr := &lifecycleManager{
|
||||
lcConfigurations: make(map[string]*proto.LcConfiguration),
|
||||
lcNodeStatus: NewLcNodeStatus(),
|
||||
lcRuleTaskStatus: NewLcRuleTaskStatus(),
|
||||
lcNodeStatus: newLcNodeStatus(),
|
||||
lcRuleTaskStatus: newLcRuleTaskStatus(),
|
||||
idleLcNodeCh: make(chan struct{}),
|
||||
exitCh: make(chan struct{}),
|
||||
}
|
||||
@ -60,8 +61,7 @@ func (lcMgr *lifecycleManager) startLcScan() {
|
||||
}
|
||||
|
||||
// start scan init
|
||||
lcMgr.lcNodeStatus.WorkingNodes = make(map[string]string)
|
||||
lcMgr.lcRuleTaskStatus = NewLcRuleTaskStatus()
|
||||
lcMgr.lcRuleTaskStatus = newLcRuleTaskStatus()
|
||||
for _, r := range tasks {
|
||||
lcMgr.lcRuleTaskStatus.ToBeScanned[r.Id] = r
|
||||
}
|
||||
@ -84,45 +84,24 @@ func (lcMgr *lifecycleManager) genEnabledRuleTasks() []*proto.RuleTask {
|
||||
}
|
||||
|
||||
func (lcMgr *lifecycleManager) scanning() bool {
|
||||
log.LogInfof("scanning lcNodeStatus: %v", lcMgr.lcNodeStatus)
|
||||
log.LogInfof("decide scanning, lcNodeStatus: %v, lcRuleTaskStatus: %v", lcMgr.lcNodeStatus, lcMgr.lcRuleTaskStatus)
|
||||
if len(lcMgr.lcRuleTaskStatus.ToBeScanned) > 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
//handling exceptions
|
||||
if len(lcMgr.lcRuleTaskStatus.Scanning) > 0 {
|
||||
scanning := lcMgr.lcRuleTaskStatus.Scanning
|
||||
var nodes []string
|
||||
lcMgr.cluster.lcNodes.Range(func(addr, value interface{}) bool {
|
||||
nodes = append(nodes, addr.(string))
|
||||
return true
|
||||
})
|
||||
|
||||
for _, task := range scanning {
|
||||
workingNodes := lcMgr.lcNodeStatus.WorkingNodes
|
||||
var node string
|
||||
for nodeAddr, t := range workingNodes {
|
||||
if task.Id == t {
|
||||
node = nodeAddr
|
||||
}
|
||||
}
|
||||
if exist(node, nodes) {
|
||||
log.LogInfof("scanning task: %v, exist node: %v, all nodes: %v", task, node, nodes)
|
||||
return true
|
||||
}
|
||||
log.LogInfof("scanning task: %v, but not exist node: %v, all nodes: %v", task, node, nodes)
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func exist(node string, nodes []string) bool {
|
||||
for _, n := range nodes {
|
||||
if node == n {
|
||||
for _, v := range lcMgr.lcRuleTaskStatus.Results {
|
||||
if v.Done != true && time.Now().Before(v.UpdateTime.Add(time.Minute*10)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
for _, c := range lcMgr.lcNodeStatus.WorkingCount {
|
||||
if c > 0 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
log.LogInfof("decide scanning, scanning stop!")
|
||||
return false
|
||||
}
|
||||
|
||||
@ -131,7 +110,7 @@ func (lcMgr *lifecycleManager) process() {
|
||||
now := time.Now()
|
||||
lcMgr.lcRuleTaskStatus.StartTime = &now
|
||||
for lcMgr.scanning() {
|
||||
log.LogDebugf("wait idleLcNodeCh... ToBeScanned num(%v), Scanning num(%v)", len(lcMgr.lcRuleTaskStatus.ToBeScanned), len(lcMgr.lcRuleTaskStatus.Scanning))
|
||||
log.LogDebugf("wait idleLcNodeCh... ToBeScanned num(%v)", len(lcMgr.lcRuleTaskStatus.ToBeScanned))
|
||||
select {
|
||||
case <-lcMgr.exitCh:
|
||||
log.LogInfo("exitCh notified, lifecycleManager process exit")
|
||||
@ -140,41 +119,35 @@ func (lcMgr *lifecycleManager) process() {
|
||||
log.LogDebug("idleLcNodeCh notified")
|
||||
|
||||
// ToBeScanned -> Scanning
|
||||
taskId := lcMgr.lcRuleTaskStatus.GetOneTask()
|
||||
if taskId == "" {
|
||||
log.LogWarn("lcRuleTaskStatus.GetOneTask, no task")
|
||||
task := lcMgr.lcRuleTaskStatus.GetOneTask()
|
||||
if task == nil {
|
||||
log.LogDebugf("lcRuleTaskStatus.GetOneTask, no task")
|
||||
continue
|
||||
}
|
||||
|
||||
// idleNodes -> workingNodes
|
||||
nodeAddr := lcMgr.lcNodeStatus.GetIdleNode(taskId)
|
||||
nodeAddr := lcMgr.lcNodeStatus.GetIdleNode()
|
||||
if nodeAddr == "" {
|
||||
log.LogWarn("no idle lcnode, redo task")
|
||||
lcMgr.lcRuleTaskStatus.RedoTask(taskId)
|
||||
lcMgr.lcRuleTaskStatus.RedoTask(task)
|
||||
continue
|
||||
}
|
||||
|
||||
val, ok := lcMgr.cluster.lcNodes.Load(nodeAddr)
|
||||
if !ok {
|
||||
log.LogErrorf("lcNodes.Load, nodeAddr(%v) is not available, redo task", nodeAddr)
|
||||
lcMgr.lcRuleTaskStatus.RedoTask(lcMgr.lcNodeStatus.RemoveNode(nodeAddr))
|
||||
lcMgr.lcNodeStatus.RemoveNode(nodeAddr)
|
||||
lcMgr.lcRuleTaskStatus.RedoTask(task)
|
||||
continue
|
||||
}
|
||||
|
||||
node := val.(*LcNode)
|
||||
task := lcMgr.lcRuleTaskStatus.GetTaskFromScanning(taskId) // task can not be nil
|
||||
if task == nil {
|
||||
log.LogErrorf("task is nil, release node(%v)", nodeAddr)
|
||||
lcMgr.lcNodeStatus.ReleaseNode(nodeAddr)
|
||||
continue
|
||||
}
|
||||
adminTask := node.createLcScanTask(lcMgr.cluster.masterAddr(), task)
|
||||
lcMgr.cluster.addLcNodeTasks([]*proto.AdminTask{adminTask})
|
||||
log.LogDebugf("add lifecycle scan task(%v) to lcnode(%v)", *task, nodeAddr)
|
||||
}
|
||||
}
|
||||
now = time.Now()
|
||||
lcMgr.lcRuleTaskStatus.EndTime = &now
|
||||
end := time.Now()
|
||||
lcMgr.lcRuleTaskStatus.EndTime = &end
|
||||
log.LogInfof("lifecycleManager process finish, lcRuleTaskStatus results(%v)", lcMgr.lcRuleTaskStatus.Results)
|
||||
}
|
||||
|
||||
@ -219,120 +192,94 @@ func (lcMgr *lifecycleManager) DelS3BucketLifecycle(VolName string) {
|
||||
//-----------------------------------------------
|
||||
|
||||
type OpLcNode interface {
|
||||
GetIdleNode(taskId string) (nodeAddr string)
|
||||
RemoveNode(nodeAddr string) (taskId string)
|
||||
ReleaseNode(nodeAddr string) (taskId string)
|
||||
GetIdleNode() (nodeAddr string)
|
||||
RemoveNode(nodeAddr string)
|
||||
UpdateNode(nodeAddr string, count int)
|
||||
}
|
||||
|
||||
// update status by heartbeat
|
||||
type lcNodeStatus struct {
|
||||
sync.RWMutex
|
||||
IdleNodes map[string]string //ip:ip
|
||||
WorkingNodes map[string]string //ip:taskId
|
||||
WorkingCount map[string]int //ip:count, number of tasks being processed on this node
|
||||
}
|
||||
|
||||
func NewLcNodeStatus() *lcNodeStatus {
|
||||
func newLcNodeStatus() *lcNodeStatus {
|
||||
return &lcNodeStatus{
|
||||
IdleNodes: make(map[string]string),
|
||||
WorkingNodes: make(map[string]string),
|
||||
WorkingCount: make(map[string]int),
|
||||
}
|
||||
}
|
||||
|
||||
func (ns *lcNodeStatus) GetIdleNode(taskId string) (nodeAddr string) {
|
||||
func (ns *lcNodeStatus) GetIdleNode() (nodeAddr string) {
|
||||
ns.Lock()
|
||||
defer ns.Unlock()
|
||||
if len(ns.IdleNodes) == 0 {
|
||||
if len(ns.WorkingCount) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
for n := range ns.IdleNodes {
|
||||
nodeAddr = n
|
||||
delete(ns.IdleNodes, nodeAddr)
|
||||
ns.WorkingNodes[nodeAddr] = taskId
|
||||
return
|
||||
var min = math.MaxInt
|
||||
for n, c := range ns.WorkingCount {
|
||||
if c <= min {
|
||||
nodeAddr = n
|
||||
min = c
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (ns *lcNodeStatus) RemoveNode(nodeAddr string) (taskId string) {
|
||||
func (ns *lcNodeStatus) RemoveNode(nodeAddr string) {
|
||||
ns.Lock()
|
||||
defer ns.Unlock()
|
||||
taskId = ns.WorkingNodes[nodeAddr]
|
||||
delete(ns.IdleNodes, nodeAddr)
|
||||
delete(ns.WorkingNodes, nodeAddr)
|
||||
delete(ns.WorkingCount, nodeAddr)
|
||||
return
|
||||
}
|
||||
|
||||
func (ns *lcNodeStatus) ReleaseNode(nodeAddr string) (taskId string) {
|
||||
func (ns *lcNodeStatus) UpdateNode(nodeAddr string, count int) {
|
||||
ns.Lock()
|
||||
defer ns.Unlock()
|
||||
taskId = ns.WorkingNodes[nodeAddr]
|
||||
delete(ns.WorkingNodes, nodeAddr)
|
||||
ns.IdleNodes[nodeAddr] = nodeAddr
|
||||
ns.WorkingCount[nodeAddr] = count
|
||||
return
|
||||
}
|
||||
|
||||
//-----------------------------------------------
|
||||
|
||||
type OpLcTask interface {
|
||||
GetOneTask() (taskId string)
|
||||
RedoTask(taskId string)
|
||||
DeleteScanningTask(taskId string)
|
||||
}
|
||||
|
||||
type lcRuleTaskStatus struct {
|
||||
sync.RWMutex
|
||||
ToBeScanned map[string]*proto.RuleTask
|
||||
Scanning map[string]*proto.RuleTask
|
||||
Results map[string]*proto.LcNodeRuleTaskResponse
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
}
|
||||
|
||||
func NewLcRuleTaskStatus() *lcRuleTaskStatus {
|
||||
func newLcRuleTaskStatus() *lcRuleTaskStatus {
|
||||
return &lcRuleTaskStatus{
|
||||
ToBeScanned: make(map[string]*proto.RuleTask),
|
||||
Scanning: make(map[string]*proto.RuleTask),
|
||||
Results: make(map[string]*proto.LcNodeRuleTaskResponse),
|
||||
}
|
||||
}
|
||||
|
||||
func (rs *lcRuleTaskStatus) GetOneTask() (taskId string) {
|
||||
func (rs *lcRuleTaskStatus) GetOneTask() (task *proto.RuleTask) {
|
||||
rs.Lock()
|
||||
defer rs.Unlock()
|
||||
if len(rs.ToBeScanned) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
for k, v := range rs.ToBeScanned {
|
||||
taskId = k
|
||||
rs.Scanning[k] = v
|
||||
delete(rs.ToBeScanned, k)
|
||||
return
|
||||
for _, t := range rs.ToBeScanned {
|
||||
task = t
|
||||
break
|
||||
}
|
||||
|
||||
delete(rs.ToBeScanned, task.Id)
|
||||
return
|
||||
}
|
||||
|
||||
func (rs *lcRuleTaskStatus) RedoTask(taskId string) {
|
||||
func (rs *lcRuleTaskStatus) RedoTask(task *proto.RuleTask) {
|
||||
rs.Lock()
|
||||
defer rs.Unlock()
|
||||
if taskId == "" {
|
||||
if task == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if task, ok := rs.Scanning[taskId]; ok {
|
||||
rs.ToBeScanned[taskId] = task
|
||||
delete(rs.Scanning, taskId)
|
||||
}
|
||||
}
|
||||
|
||||
func (rs *lcRuleTaskStatus) DeleteScanningTask(taskId string) {
|
||||
rs.Lock()
|
||||
defer rs.Unlock()
|
||||
if taskId == "" {
|
||||
return
|
||||
}
|
||||
|
||||
delete(rs.Scanning, taskId)
|
||||
rs.ToBeScanned[task.Id] = task
|
||||
}
|
||||
|
||||
func (rs *lcRuleTaskStatus) AddResult(resp *proto.LcNodeRuleTaskResponse) {
|
||||
@ -340,9 +287,3 @@ func (rs *lcRuleTaskStatus) AddResult(resp *proto.LcNodeRuleTaskResponse) {
|
||||
defer rs.Unlock()
|
||||
rs.Results[resp.ID] = resp
|
||||
}
|
||||
|
||||
func (rs *lcRuleTaskStatus) GetTaskFromScanning(taskId string) *proto.RuleTask {
|
||||
rs.Lock()
|
||||
defer rs.Unlock()
|
||||
return rs.Scanning[taskId]
|
||||
}
|
||||
|
||||
@ -87,60 +87,50 @@ func (c *Cluster) handleLcNodeHeartbeatResp(nodeAddr string, resp *proto.LcNodeH
|
||||
lcNode.ReportTime = time.Now()
|
||||
lcNode.Unlock()
|
||||
|
||||
//update lcNodeStatus
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v], LcScanningTasks[%v], SnapshotScanningTasks[%v]", nodeAddr, len(resp.LcScanningTasks), len(resp.SnapshotScanningTasks))
|
||||
c.lcMgr.lcNodeStatus.UpdateNode(nodeAddr, len(resp.LcScanningTasks))
|
||||
c.snapshotMgr.lcNodeStatus.UpdateNode(nodeAddr, len(resp.SnapshotScanningTasks))
|
||||
|
||||
//handle LcScanningTasks
|
||||
if len(resp.LcScanningTasks) != 0 {
|
||||
for _, taskRsp := range resp.LcScanningTasks {
|
||||
c.lcMgr.lcRuleTaskStatus.Lock()
|
||||
for _, taskRsp := range resp.LcScanningTasks {
|
||||
c.lcMgr.lcRuleTaskStatus.Lock()
|
||||
|
||||
if c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID] != nil && c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID].Done {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] task[%v] already done", nodeAddr, taskRsp.ID)
|
||||
} else {
|
||||
c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID] = taskRsp
|
||||
}
|
||||
|
||||
c.lcMgr.lcRuleTaskStatus.Unlock()
|
||||
log.LogDebugf("action[handleLcNodeHeartbeatResp], lcNode[%v] taskRsp: %v", nodeAddr, taskRsp)
|
||||
//avoid updating TaskResults incorrectly when received handleLcNodeLcScanResp first and then handleLcNodeHeartbeatResp
|
||||
if c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID] != nil && c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID].Done {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] task[%v] already done", nodeAddr, taskRsp.ID)
|
||||
} else {
|
||||
t := time.Now()
|
||||
taskRsp.UpdateTime = &t
|
||||
c.lcMgr.lcRuleTaskStatus.Results[taskRsp.ID] = taskRsp
|
||||
}
|
||||
} else {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] is idle for LcScanningTasks", nodeAddr)
|
||||
c.lcMgr.lcRuleTaskStatus.DeleteScanningTask(c.lcMgr.lcNodeStatus.ReleaseNode(nodeAddr))
|
||||
|
||||
c.lcMgr.lcRuleTaskStatus.Unlock()
|
||||
log.LogDebugf("action[handleLcNodeHeartbeatResp], lcNode[%v] taskRsp: %v", nodeAddr, taskRsp)
|
||||
}
|
||||
if len(resp.LcScanningTasks) < resp.LcTaskCountLimit {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], notify idle lcNode[%v], now LcScanningTasks[%v]", nodeAddr, len(resp.LcScanningTasks))
|
||||
c.lcMgr.notifyIdleLcNode()
|
||||
}
|
||||
|
||||
//handle SnapshotScanningTasks
|
||||
if len(resp.SnapshotScanningTasks) != 0 {
|
||||
for _, taskRsp := range resp.SnapshotScanningTasks {
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.Lock()
|
||||
for _, taskRsp := range resp.SnapshotScanningTasks {
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.Lock()
|
||||
|
||||
if c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID] != nil && c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID].Done {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot task[%v] already done", nodeAddr, taskRsp.ID)
|
||||
} else {
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID] = taskRsp
|
||||
|
||||
//update processing status
|
||||
if pInfo, ok := c.snapshotMgr.lcSnapshotTaskStatus.ProcessingVerInfos[taskRsp.ID]; ok {
|
||||
pInfo.UpdateTime = time.Now().Unix()
|
||||
log.LogDebugf("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot task[%v] update time", nodeAddr, taskRsp.ID)
|
||||
} else {
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.ProcessingVerInfos[taskRsp.ID] = &proto.SnapshotVerDelTask{
|
||||
Id: fmt.Sprintf("%s:%d", taskRsp.VolName, taskRsp.VerSeq),
|
||||
VolName: taskRsp.VolName,
|
||||
UpdateTime: time.Now().Unix(),
|
||||
VolVersionInfo: &proto.VolVersionInfo{
|
||||
Ver: taskRsp.VerSeq,
|
||||
Status: proto.VersionDeleting,
|
||||
},
|
||||
}
|
||||
log.LogWarnf("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot task[%v] not in processing add it", nodeAddr, taskRsp.ID)
|
||||
}
|
||||
}
|
||||
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.Unlock()
|
||||
log.LogDebugf("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot taskRsp: %v", nodeAddr, taskRsp)
|
||||
//avoid updating TaskResults incorrectly when received handleLcNodeLcScanResp first and then handleLcNodeHeartbeatResp
|
||||
if c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID] != nil && c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID].Done {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot task[%v] already done", nodeAddr, taskRsp.ID)
|
||||
} else {
|
||||
t := time.Now()
|
||||
taskRsp.UpdateTime = &t
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.TaskResults[taskRsp.ID] = taskRsp
|
||||
}
|
||||
} else {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], lcNode[%v] is idle for SnapshotScanningTasks", nodeAddr)
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.DeleteScanningTask(c.snapshotMgr.lcNodeStatus.ReleaseNode(nodeAddr))
|
||||
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.Unlock()
|
||||
log.LogDebugf("action[handleLcNodeHeartbeatResp], lcNode[%v] snapshot taskRsp: %v", nodeAddr, taskRsp)
|
||||
}
|
||||
if len(resp.SnapshotScanningTasks) < resp.LcTaskCountLimit {
|
||||
log.LogInfof("action[handleLcNodeHeartbeatResp], notify idle lcNode[%v], now SnapshotScanningTasks[%v]", nodeAddr, len(resp.SnapshotScanningTasks))
|
||||
c.snapshotMgr.notifyIdleLcNode()
|
||||
}
|
||||
|
||||
@ -156,17 +146,16 @@ func (c *Cluster) handleLcNodeLcScanResp(nodeAddr string, resp *proto.LcNodeRule
|
||||
|
||||
switch resp.Status {
|
||||
case proto.TaskFailed:
|
||||
c.lcMgr.lcRuleTaskStatus.RedoTask(resp.ID)
|
||||
log.LogWarnf("action[handleLcNodeLcScanResp] scanning failed, resp(%v), no redo", resp)
|
||||
return
|
||||
case proto.TaskSucceeds:
|
||||
c.lcMgr.lcRuleTaskStatus.AddResult(resp)
|
||||
c.lcMgr.lcRuleTaskStatus.DeleteScanningTask(resp.ID)
|
||||
log.LogInfof("action[handleLcNodeLcScanResp] scanning completed, resp(%v)", resp)
|
||||
return
|
||||
default:
|
||||
log.LogInfof("action[handleLcNodeLcScanResp] scanning received, resp(%v)", resp)
|
||||
return
|
||||
}
|
||||
|
||||
c.lcMgr.lcNodeStatus.ReleaseNode(nodeAddr)
|
||||
log.LogInfof("action[handleLcNodeLcScanResp] scanning completed, resp(%v)", resp)
|
||||
return
|
||||
}
|
||||
|
||||
@ -178,7 +167,9 @@ func (c *Cluster) handleLcNodeSnapshotScanResp(nodeAddr string, resp *proto.Snap
|
||||
|
||||
switch resp.Status {
|
||||
case proto.TaskFailed:
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.RedoTask(resp.ID)
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.RedoTask(resp.SnapshotVerDelTask)
|
||||
log.LogWarnf("action[handleLcNodeSnapshotScanResp] scanning failed, resp(%v), redo", resp)
|
||||
return
|
||||
case proto.TaskSucceeds:
|
||||
//1.mark done for VersionMgr
|
||||
var vol *Vol
|
||||
@ -192,13 +183,11 @@ func (c *Cluster) handleLcNodeSnapshotScanResp(nodeAddr string, resp *proto.Snap
|
||||
|
||||
//2. mark done for snapshotMgr
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.AddResult(resp)
|
||||
c.snapshotMgr.lcSnapshotTaskStatus.DeleteScanningTask(resp.ID)
|
||||
log.LogInfof("action[handleLcNodeSnapshotScanResp] scanning completed, resp(%v)", resp)
|
||||
return
|
||||
default:
|
||||
log.LogInfof("action[handleLcNodeSnapshotScanResp] scanning received, resp(%v)", resp)
|
||||
return
|
||||
}
|
||||
|
||||
c.snapshotMgr.lcNodeStatus.ReleaseNode(nodeAddr)
|
||||
log.LogInfof("action[handleLcNodeSnapshotScanResp] scanning completed, resp(%v)", resp)
|
||||
return
|
||||
}
|
||||
|
||||
@ -34,8 +34,8 @@ type snapshotDelManager struct {
|
||||
func newSnapshotManager() *snapshotDelManager {
|
||||
log.LogInfof("action[newSnapshotManager] construct")
|
||||
snapshotMgr := &snapshotDelManager{
|
||||
lcSnapshotTaskStatus: NewLcSnapshotVerStatus(),
|
||||
lcNodeStatus: NewLcNodeStatus(),
|
||||
lcSnapshotTaskStatus: newLcSnapshotVerStatus(),
|
||||
lcNodeStatus: newLcNodeStatus(),
|
||||
idleNodeCh: make(chan struct{}),
|
||||
exitCh: make(chan struct{}),
|
||||
}
|
||||
@ -51,33 +51,28 @@ func (m *snapshotDelManager) process() {
|
||||
case <-m.idleNodeCh:
|
||||
log.LogDebug("idleLcNodeCh notified")
|
||||
|
||||
taskId := m.lcSnapshotTaskStatus.GetOneTask()
|
||||
if taskId == "" {
|
||||
log.LogDebugf("no snapshot ver del task available")
|
||||
task := m.lcSnapshotTaskStatus.GetOneTask()
|
||||
if task == nil {
|
||||
log.LogDebugf("lcSnapshotTaskStatus.GetOneTask, no task")
|
||||
continue
|
||||
}
|
||||
|
||||
nodeAddr := m.lcNodeStatus.GetIdleNode(taskId)
|
||||
nodeAddr := m.lcNodeStatus.GetIdleNode()
|
||||
if nodeAddr == "" {
|
||||
log.LogWarn("no idle lcnode, redo task")
|
||||
m.lcSnapshotTaskStatus.RedoTask(taskId)
|
||||
m.lcSnapshotTaskStatus.RedoTask(task)
|
||||
continue
|
||||
}
|
||||
|
||||
val, ok := m.cluster.lcNodes.Load(nodeAddr)
|
||||
if !ok {
|
||||
log.LogErrorf("lcNodes.Load, nodeAddr(%v) is not available, redo task", nodeAddr)
|
||||
m.lcSnapshotTaskStatus.RedoTask(m.lcNodeStatus.RemoveNode(nodeAddr))
|
||||
m.lcNodeStatus.RemoveNode(nodeAddr)
|
||||
m.lcSnapshotTaskStatus.RedoTask(task)
|
||||
continue
|
||||
}
|
||||
|
||||
node := val.(*LcNode)
|
||||
task := m.lcSnapshotTaskStatus.GetTaskFromScanning(taskId)
|
||||
if task == nil {
|
||||
log.LogErrorf("task is nil, release node(%v)", nodeAddr)
|
||||
m.lcNodeStatus.ReleaseNode(nodeAddr)
|
||||
continue
|
||||
}
|
||||
adminTask := node.createSnapshotVerDelTask(m.cluster.masterAddr(), task)
|
||||
m.cluster.addLcNodeTasks([]*proto.AdminTask{adminTask})
|
||||
log.LogDebugf("add snapshot version del task(%v) to lcnode(%v)", *task, nodeAddr)
|
||||
@ -103,20 +98,18 @@ func (m *snapshotDelManager) notifyIdleLcNode() {
|
||||
|
||||
type lcSnapshotVerStatus struct {
|
||||
sync.RWMutex
|
||||
VerInfos map[string]*proto.SnapshotVerDelTask
|
||||
ProcessingVerInfos map[string]*proto.SnapshotVerDelTask
|
||||
TaskResults map[string]*proto.SnapshotVerDelTaskResponse
|
||||
VerInfos map[string]*proto.SnapshotVerDelTask
|
||||
TaskResults map[string]*proto.SnapshotVerDelTaskResponse
|
||||
}
|
||||
|
||||
func NewLcSnapshotVerStatus() *lcSnapshotVerStatus {
|
||||
func newLcSnapshotVerStatus() *lcSnapshotVerStatus {
|
||||
return &lcSnapshotVerStatus{
|
||||
VerInfos: make(map[string]*proto.SnapshotVerDelTask, 0),
|
||||
ProcessingVerInfos: make(map[string]*proto.SnapshotVerDelTask, 0),
|
||||
TaskResults: make(map[string]*proto.SnapshotVerDelTaskResponse, 0),
|
||||
VerInfos: make(map[string]*proto.SnapshotVerDelTask, 0),
|
||||
TaskResults: make(map[string]*proto.SnapshotVerDelTaskResponse, 0),
|
||||
}
|
||||
}
|
||||
|
||||
func (vs *lcSnapshotVerStatus) GetOneTask() (taskId string) {
|
||||
func (vs *lcSnapshotVerStatus) GetOneTask() (task *proto.SnapshotVerDelTask) {
|
||||
var min int64 = math.MaxInt64
|
||||
|
||||
vs.Lock()
|
||||
@ -128,52 +121,32 @@ func (vs *lcSnapshotVerStatus) GetOneTask() (taskId string) {
|
||||
for _, i := range vs.VerInfos {
|
||||
if i.VolVersionInfo.DelTime < min {
|
||||
min = i.VolVersionInfo.DelTime
|
||||
taskId = i.Id
|
||||
task = i
|
||||
}
|
||||
}
|
||||
|
||||
info := vs.VerInfos[taskId]
|
||||
delete(vs.VerInfos, taskId)
|
||||
info.UpdateTime = time.Now().Unix()
|
||||
vs.ProcessingVerInfos[taskId] = info
|
||||
delete(vs.VerInfos, task.Id)
|
||||
return
|
||||
}
|
||||
|
||||
func (vs *lcSnapshotVerStatus) RedoTask(taskId string) {
|
||||
func (vs *lcSnapshotVerStatus) RedoTask(task *proto.SnapshotVerDelTask) {
|
||||
vs.Lock()
|
||||
defer vs.Unlock()
|
||||
if taskId == "" {
|
||||
if task == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if pInfo, ok := vs.ProcessingVerInfos[taskId]; ok {
|
||||
vs.VerInfos[taskId] = pInfo
|
||||
delete(vs.ProcessingVerInfos, taskId)
|
||||
}
|
||||
}
|
||||
|
||||
func (vs *lcSnapshotVerStatus) DeleteScanningTask(taskId string) {
|
||||
vs.Lock()
|
||||
defer vs.Unlock()
|
||||
if taskId == "" {
|
||||
return
|
||||
}
|
||||
|
||||
delete(vs.ProcessingVerInfos, taskId)
|
||||
vs.VerInfos[task.Id] = task
|
||||
}
|
||||
|
||||
func (vs *lcSnapshotVerStatus) AddVerInfo(task *proto.SnapshotVerDelTask) {
|
||||
vs.Lock()
|
||||
defer vs.Unlock()
|
||||
if pInfo, ok := vs.ProcessingVerInfos[task.Id]; ok {
|
||||
if time.Since(time.Unix(pInfo.UpdateTime, 0)) < time.Second*time.Duration(5*defaultIntervalToCheck) {
|
||||
log.LogDebugf("VerInfo: %v is already in processing", task)
|
||||
return
|
||||
} else {
|
||||
log.LogWarnf("VerInfo: %v is in processing, but expired", task)
|
||||
}
|
||||
}
|
||||
|
||||
if _, ok := vs.TaskResults[task.Id]; ok {
|
||||
log.LogDebugf("VerInfo: %v is in TaskResults, already in processing", task)
|
||||
return
|
||||
}
|
||||
vs.VerInfos[task.Id] = task
|
||||
log.LogDebugf("Add VerInfo task: %v", task)
|
||||
}
|
||||
@ -188,15 +161,15 @@ func (vs *lcSnapshotVerStatus) DeleteOldResult() {
|
||||
vs.Lock()
|
||||
defer vs.Unlock()
|
||||
for k, v := range vs.TaskResults {
|
||||
if v.Done == true && time.Now().After(v.EndTime.Add(time.Hour*1)) {
|
||||
// delete result that already done
|
||||
if v.Done == true && time.Now().After(v.EndTime.Add(time.Minute*10)) {
|
||||
delete(vs.TaskResults, k)
|
||||
log.LogDebugf("delete old result: %v", v)
|
||||
log.LogDebugf("delete result already done: %v", v)
|
||||
}
|
||||
// delete result that not done but no updating
|
||||
if v.Done != true && time.Now().After(v.UpdateTime.Add(time.Minute*10)) {
|
||||
delete(vs.TaskResults, k)
|
||||
log.LogWarnf("delete result that not done but no updating: %v", v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (vs *lcSnapshotVerStatus) GetTaskFromScanning(taskId string) *proto.SnapshotVerDelTask {
|
||||
vs.Lock()
|
||||
defer vs.Unlock()
|
||||
return vs.ProcessingVerInfos[taskId]
|
||||
}
|
||||
|
||||
@ -689,6 +689,7 @@ type MetaNodeHeartbeatResponse struct {
|
||||
type LcNodeHeartbeatResponse struct {
|
||||
Status uint8
|
||||
Result string
|
||||
LcTaskCountLimit int
|
||||
LcScanningTasks map[string]*LcNodeRuleTaskResponse
|
||||
SnapshotScanningTasks map[string]*SnapshotVerDelTaskResponse
|
||||
}
|
||||
|
||||
@ -83,12 +83,14 @@ type RuleTask struct {
|
||||
}
|
||||
|
||||
type LcNodeRuleTaskResponse struct {
|
||||
ID string
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
Done bool
|
||||
Status uint8
|
||||
Result string
|
||||
ID string
|
||||
LcNode string
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
UpdateTime *time.Time
|
||||
Done bool
|
||||
Status uint8
|
||||
Result string
|
||||
LcNodeRuleTaskStatistics
|
||||
}
|
||||
|
||||
|
||||
@ -13,17 +13,19 @@ type SnapshotVerDelTaskRequest struct {
|
||||
type SnapshotVerDelTask struct {
|
||||
Id string
|
||||
VolName string
|
||||
UpdateTime int64
|
||||
VolVersionInfo *VolVersionInfo
|
||||
}
|
||||
|
||||
type SnapshotVerDelTaskResponse struct {
|
||||
ID string
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
Done bool
|
||||
Status uint8
|
||||
Result string
|
||||
ID string
|
||||
LcNode string
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
UpdateTime *time.Time
|
||||
Done bool
|
||||
Status uint8
|
||||
Result string
|
||||
SnapshotVerDelTask *SnapshotVerDelTask
|
||||
SnapshotStatistics
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
