-
Notifications
You must be signed in to change notification settings - Fork 1
/
scheduler.go
82 lines (73 loc) · 2.28 KB
/
scheduler.go
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
package run
import (
"time"
"github.com/utilitywarehouse/terraform-applier/git"
"github.com/utilitywarehouse/terraform-applier/log"
)
// Scheduler handles queueing apply runs at a given time interval and upon every new Git commit.
type Scheduler struct {
Errors chan<- error
FullRunInterval time.Duration
GitUtil git.UtilInterface
PollInterval time.Duration
ModulesPathFilters []string
RunQueue chan<- bool
}
// Start runs a continuous loop with two tickers for queueing runs.
// One ticker queues a new run every X seconds, where X is the value from $FULL_RUN_INTERVAL_SECONDS.
// The other ticker queues a new run upon every new Git commit, checking the repo every Y seconds where Y is the value from $POLL_INTERVAL_SECONDS.
func (s *Scheduler) Start() {
var fullRunTickerChan, pollTickerChan <-chan time.Time
if s.FullRunInterval != 0 {
fullRunTicker := time.NewTicker(s.FullRunInterval)
defer fullRunTicker.Stop()
fullRunTickerChan = fullRunTicker.C
}
var lastCommitHash string
isRepo, err := s.GitUtil.IsRepo()
if err != nil {
s.Errors <- err
return
}
if isRepo {
pollTicker := time.NewTicker(s.PollInterval)
defer pollTicker.Stop()
pollTickerChan = pollTicker.C
lastCommitHash, err = s.GitUtil.HeadHashForPaths(s.ModulesPathFilters...)
if err != nil {
s.Errors <- err
return
}
}
log.Info("Queueing first run")
s.enqueue(s.RunQueue)
for {
select {
case <-fullRunTickerChan:
log.Info("Full run interval reached, queueing run, interval=%v", s.FullRunInterval)
s.enqueue(s.RunQueue)
case <-pollTickerChan:
newCommitHash, err := s.GitUtil.HeadHashForPaths(s.ModulesPathFilters...)
if err != nil {
s.Errors <- err
return
}
if newCommitHash != lastCommitHash {
log.Info("Queueing run, newest-commit=%s, last-commit=%s", newCommitHash, lastCommitHash)
s.enqueue(s.RunQueue)
lastCommitHash = newCommitHash
} else {
log.Debug("Run not queued, no new commit, newest-commit=%s, last-commit=%s", newCommitHash, lastCommitHash)
}
}
}
}
// enqueue attempts to add a run to the queue, logging the result of the request.
func (s *Scheduler) enqueue(runQueue chan<- bool) {
select {
case runQueue <- true:
log.Info("Run queued")
default:
log.Info("Run queue is already full")
}
}