forked from coreos/fleet
-
Notifications
You must be signed in to change notification settings - Fork 0
/
agent.go
142 lines (115 loc) · 3.07 KB
/
agent.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
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
package agent
import (
"encoding/json"
"fmt"
"time"
"github.com/coreos/fleet/job"
"github.com/coreos/fleet/log"
"github.com/coreos/fleet/machine"
"github.com/coreos/fleet/pkg"
"github.com/coreos/fleet/registry"
"github.com/coreos/fleet/unit"
)
const (
// TTL to use with all state pushed to Registry
DefaultTTL = "30s"
)
type Agent struct {
registry registry.Registry
um unit.UnitManager
uGen *unit.UnitStateGenerator
Machine machine.Machine
ttl time.Duration
cache *agentCache
}
func New(mgr unit.UnitManager, uGen *unit.UnitStateGenerator, reg registry.Registry, mach machine.Machine, ttl time.Duration) *Agent {
return &Agent{reg, mgr, uGen, mach, ttl, &agentCache{}}
}
func (a *Agent) MarshalJSON() ([]byte, error) {
data := struct {
Cache *agentCache
}{
Cache: a.cache,
}
return json.Marshal(data)
}
// Heartbeat updates the Registry periodically with an acknowledgement of the
// Jobs this Agent is expected to be running.
func (a *Agent) Heartbeat(stop chan bool) {
a.heartbeatJobs(a.ttl, stop)
}
func (a *Agent) heartbeatJobs(ttl time.Duration, stop chan bool) {
heartbeat := func() {
machID := a.Machine.State().ID
launched := a.cache.launchedJobs()
for _, j := range launched {
go a.registry.UnitHeartbeat(j, machID, ttl)
}
}
interval := ttl / 2
ticker := time.Tick(interval)
for {
select {
case <-stop:
log.V(1).Info("HeartbeatJobs exiting due to stop signal")
return
case <-ticker:
log.V(1).Info("HeartbeatJobs tick")
heartbeat()
}
}
}
func (a *Agent) loadUnit(u *job.Unit) error {
a.cache.setTargetState(u.Name, job.JobStateLoaded)
a.uGen.Subscribe(u.Name)
return a.um.Load(u.Name, u.Unit)
}
func (a *Agent) unloadUnit(unitName string) {
a.registry.ClearUnitHeartbeat(unitName)
a.cache.dropTargetState(unitName)
a.um.TriggerStop(unitName)
a.uGen.Unsubscribe(unitName)
a.um.Unload(unitName)
}
func (a *Agent) startUnit(unitName string) {
a.cache.setTargetState(unitName, job.JobStateLaunched)
machID := a.Machine.State().ID
a.registry.UnitHeartbeat(unitName, machID, a.ttl)
a.um.TriggerStart(unitName)
}
func (a *Agent) stopUnit(unitName string) {
a.cache.setTargetState(unitName, job.JobStateLoaded)
a.registry.ClearUnitHeartbeat(unitName)
a.um.TriggerStop(unitName)
}
type unitStates map[string]job.JobState
// units returns a map representing the current state of units known by the agent.
func (a *Agent) units() (unitStates, error) {
launched := pkg.NewUnsafeSet()
for _, jName := range a.cache.launchedJobs() {
launched.Add(jName)
}
loaded := pkg.NewUnsafeSet()
for _, jName := range a.cache.loadedJobs() {
loaded.Add(jName)
}
units, err := a.um.Units()
if err != nil {
return nil, fmt.Errorf("failed fetching loaded units from UnitManager: %v", err)
}
filter := pkg.NewUnsafeSet()
for _, u := range units {
filter.Add(u)
}
states := make(unitStates)
for _, uName := range units {
js := job.JobStateInactive
if loaded.Contains(uName) {
js = job.JobStateLoaded
} else if launched.Contains(uName) {
js = job.JobStateLaunched
}
states[uName] = js
}
return states, nil
}