forked from coreos/fleet
-
Notifications
You must be signed in to change notification settings - Fork 0
/
offer.go
152 lines (126 loc) · 3.01 KB
/
offer.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
143
144
145
146
147
148
149
150
151
152
package registry
import (
"path"
"strings"
log "github.com/coreos/fleet/Godeps/_workspace/src/github.com/golang/glog"
"github.com/coreos/fleet/etcd"
"github.com/coreos/fleet/job"
"github.com/coreos/fleet/pkg"
)
const (
offerPrefix = "offer"
)
// jobOfferModel is used for serializing and deserializing JobOffers stored in the Registry
type jobOfferModel struct {
Job jobModel
MachineIDs []string
}
// CreateJobOffer attempts to store a JobOffer and a reference to its associated Job in the repository
func (r *EtcdRegistry) CreateJobOffer(jo *job.JobOffer) error {
jom := jobOfferModel{
Job: jobModel{
Name: jo.Job.Name,
UnitHash: jo.Job.Unit.Hash(),
},
MachineIDs: jo.MachineIDs,
}
json, err := marshal(jom)
if err != nil {
log.Errorf(err.Error())
return err
}
req := etcd.Set{
Key: path.Join(r.keyPrefix, offerPrefix, jo.Job.Name, "object"),
Value: json,
}
_, err = r.etcd.Do(&req)
return err
}
// getJobOfferFromJSON hydrates a JobOffer from a JSON-encoded jobOfferModel
func (r *EtcdRegistry) getJobOfferFromJSON(val string) *job.JobOffer {
var jom jobOfferModel
if err := unmarshal(val, &jom); err != nil {
return nil
}
j := r.getJobFromModel(jom.Job)
if j == nil {
return nil
}
jo := job.JobOffer{
Job: *j,
MachineIDs: jom.MachineIDs,
}
return &jo
}
// Bids returns a list of machine IDs that have bid for the referenced Job
func (r *EtcdRegistry) Bids(jName string) (bids pkg.Set, err error) {
bids = pkg.NewUnsafeSet()
req := etcd.Get{
Key: path.Join(r.keyPrefix, offerPrefix, jName, "bids"),
Recursive: true,
}
var resp *etcd.Result
resp, err = r.etcd.Do(&req)
if err != nil {
if isKeyNotFound(err) {
err = nil
}
return
}
for _, node := range resp.Node.Nodes {
machID := path.Base(node.Key)
bids.Add(machID)
}
return
}
// UnresolvedJobOffers returns a list of hydrated JobOffers from the Registry
func (r *EtcdRegistry) UnresolvedJobOffers() ([]job.JobOffer, error) {
req := etcd.Get{
Key: path.Join(r.keyPrefix, offerPrefix),
Sorted: true,
Recursive: true,
}
resp, err := r.etcd.Do(&req)
if err != nil {
if isKeyNotFound(err) {
err = nil
}
return nil, err
}
var offers []job.JobOffer
for _, node := range resp.Node.Nodes {
for _, obj := range node.Nodes {
if !strings.HasSuffix(obj.Key, "/object") {
continue
}
jo := r.getJobOfferFromJSON(obj.Value)
if jo == nil {
continue
}
offers = append(offers, *jo)
}
}
return offers, nil
}
func (r *EtcdRegistry) ResolveJobOffer(jobName string) error {
req := etcd.Delete{
Key: path.Join(r.keyPrefix, offerPrefix, jobName, "object"),
}
if _, err := r.etcd.Do(&req); err != nil {
if !isKeyNotFound(err) {
return err
}
}
req = etcd.Delete{
Key: path.Join(r.keyPrefix, offerPrefix, jobName),
Recursive: true,
}
r.etcd.Do(&req)
return nil
}
func (r *EtcdRegistry) SubmitJobBid(jName, machID string) {
req := etcd.Set{
Key: path.Join(r.keyPrefix, offerPrefix, jName, "bids", machID),
}
r.etcd.Do(&req)
}