flare-p-chain-indexer/indexer/cronjob/cronjob.go at 033ee49a8fc8a87f2241ef40ad7773efa55f93df · flare-foundation/flare-p-chain-indexer · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
package cronjob

import (
	"flare-indexer/indexer/config"
	"flare-indexer/indexer/shared"
	"flare-indexer/logger"
	"flare-indexer/utils"
	"flare-indexer/utils/staking"
	"time"
)

type Cronjob interface {
	Name() string
	Enabled() bool
	Timeout() time.Duration
	RandomTimeoutDelta() time.Duration
	Call() error
	OnStart() error

	// Set health status of cronjob
	// (can be implemented to ignore the status based on other conditions)
	UpdateCronjobStatus(status shared.HealthStatus)
}

func RunCronjob(c Cronjob) {
	if !c.Enabled() {
		logger.Debug("%s cronjob disabled", c.Name())
		c.UpdateCronjobStatus(shared.HealthStatusOk)
		return
	}

	err := c.OnStart()
	if err != nil {
		logger.Error("%s cronjob on start error %v", c.Name(), err)
		return
	}

	logger.Debug("starting %s cronjob", c.Name())

	ticker := utils.NewRandomizedTicker(c.Timeout(), c.RandomTimeoutDelta())
	for {
		<-ticker

		err := c.Call()
		if err == nil {
			c.UpdateCronjobStatus(shared.HealthStatusOk)
		} else {
			logger.Error("%s cronjob error %s", c.Name(), err.Error())
			c.UpdateCronjobStatus(shared.HealthStatusError)
		}
	}
}

const (
	defaultEpochBatchSize int64 = 100
)

type epochCronjob struct {
	enabled   bool
	timeout   time.Duration // call cronjob every "timeout"
	epochs    staking.EpochInfo
	delay     time.Duration // voting delay
	batchSize int64
	metrics   *epochCronjobMetrics
}

type epochRange struct {
	start int64
	end   int64
}

func newEpochCronjob(cronjobCfg *config.CronjobConfig, epochs staking.EpochInfo) epochCronjob {
	return epochCronjob{
		enabled:   cronjobCfg.Enabled,
		timeout:   cronjobCfg.Timeout,
		epochs:    epochs,
		batchSize: cronjobCfg.BatchSize,
		delay:     cronjobCfg.Delay,
	}
}

func (c *epochCronjob) Enabled() bool {
	return c.enabled
}

func (c *epochCronjob) Timeout() time.Duration {
	return c.timeout
}

func (c *epochCronjob) RandomTimeoutDelta() time.Duration {
	return 0
}

func (c *epochCronjob) UpdateCronjobStatus(status shared.HealthStatus) {
	if c.metrics != nil {
		c.metrics.SetStatus(status)
	}
}

// Get trimmed processing range (closed interval)
func (c *epochCronjob) getTrimmedEpochRange(start, end int64) *epochRange {
	start = utils.Max(start, c.epochs.First)
	batchSize := c.batchSize
	if batchSize == 0 {
		batchSize = defaultEpochBatchSize
	} else if batchSize < 0 {
		batchSize = end - start + 1
	}
	if end >= start+batchSize {
		end = batchSize + start - 1
	}
	return &epochRange{start, end}
}

func (c *epochCronjob) updateLastEpochMetrics(epoch int64) {
	if c.metrics != nil {
		c.metrics.lastEpoch.Set(float64(epoch))
	}
}

func (c *epochCronjob) updateLastProcessedEpochMetrics(epoch int64) {
	if c.metrics != nil {
		c.metrics.lastProcessedEpoch.Set(float64(epoch))
	}
}