cubefs

1
package datanode
2

3
import (
4
	"sync/atomic"
5
	"time"
6

7
	"github.com/cubefs/cubefs/util/log"
8
	"golang.org/x/time/rate"
9
)
10

11
const (
12
	defaultMarkDeleteLimitRate  = rate.Inf
13
	defaultMarkDeleteLimitBurst = 512
14
	defaultIOLimitBurst         = 512
15
	UpdateNodeInfoTicket        = 1 * time.Minute
16

17
	RepairTimeOut   = time.Hour * 24
18
	MaxRepairErrCnt = 1000
19
)
20

21
var nodeInfoStopC = make(chan struct{})
22

23
func (m *DataNode) startUpdateNodeInfo() {
24
	ticker := time.NewTicker(UpdateNodeInfoTicket)
25
	defer ticker.Stop()
26
	for {
27
		select {
28
		case <-nodeInfoStopC:
29
			log.LogInfo("datanode nodeinfo goroutine stopped")
30
			return
31
		case <-ticker.C:
32
			m.updateNodeInfo()
33
		}
34
	}
35
}
36

37
func (m *DataNode) stopUpdateNodeInfo() {
38
	nodeInfoStopC <- struct{}{}
39
}
40

41
func (m *DataNode) updateNodeInfo() {
42
	clusterInfo, err := MasterClient.AdminAPI().GetClusterInfo()
43
	if err != nil {
44
		log.LogErrorf("[updateDataNodeInfo] %s", err.Error())
45
		return
46
	}
47

48
	setLimiter(deleteLimiteRater, clusterInfo.DataNodeDeleteLimitRate)
49

50
	setDoExtentRepair(int(clusterInfo.DataNodeAutoRepairLimitRate))
51

52
	atomic.StoreUint64(&m.dpMaxRepairErrCnt, clusterInfo.DpMaxRepairErrCnt)
53

54
	log.LogInfof("updateNodeInfo from master:"+
55
		"deleteLimite(%v), autoRepairLimit(%v), dpMaxRepairErrCnt(%v)",
56
		clusterInfo.DataNodeDeleteLimitRate, clusterInfo.DataNodeAutoRepairLimitRate,
57
		clusterInfo.DpMaxRepairErrCnt)
58
}
59

60
func (m *DataNode) GetDpMaxRepairErrCnt() uint64 {
61
	dpMaxRepairErrCnt := atomic.LoadUint64(&m.dpMaxRepairErrCnt)
62
	if dpMaxRepairErrCnt == 0 {
63
		return MaxRepairErrCnt
64
	}
65
	return dpMaxRepairErrCnt
66
}
67