amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostops
2
3import (
4 "context"
5 "encoding/json"
6 "math"
7 "reflect"
8 "time"
9
10 "github.com/carlosframework/platform/internal/controlname"
11 "github.com/carlosframework/platform/internal/hostclaim"
12 "github.com/carlosframework/platform/internal/release"
13)
14
15const RegistrationTTL = 90 * time.Second
16const RegistrationRenewEvery = 20 * time.Second
17const RegistrationSkew = 5 * time.Second
18
19type RegistrationOwner struct {
20 Console string `json:"console,omitempty"`
21 Box string `json:"box,omitempty"`
22 Enrollment string `json:"enrollment"`
23 Controller string `json:"controller"`
24}
25type SavedClaim struct {
26 Absent bool `json:"absent"`
27 ETag string `json:"etag,omitempty"`
28 Gen uint64 `json:"gen,omitempty"`
29}
30type Registration struct {
31 Kind string `json:"kind"`
32 Host string `json:"host"`
33 Owner RegistrationOwner `json:"owner"`
34 AppID string `json:"app_id"`
35 Epoch uint64 `json:"epoch"`
36 Claim SavedClaim `json:"claim"`
37 Expires time.Time `json:"expires"`
38 CreatedAt time.Time `json:"created_at"`
39 Incarnation string `json:"incarnation"`
40}
41type Effect struct {
42 ID string `json:"id"`
43 OperationID string `json:"operation_id"`
44 AppID string `json:"app_id"`
45 Resource string `json:"resource"`
46 Action string `json:"action"`
47 Inputs map[string]string `json:"inputs"`
48 Owner RegistrationOwner `json:"owner"`
49 State string `json:"state"`
50 ResultReference string `json:"result_reference,omitempty"`
51}
52type AppOperation struct {
53 ID string `json:"id"`
54 Kind string `json:"kind"`
55 AppID string `json:"app_id"`
56 Phase string `json:"phase"`
57 Intent map[string]string `json:"intent"`
58}
59type PurgeMark struct {
60 ID string `json:"id"`
61 AppID string `json:"app_id"`
62}
63type RestoreMark struct {
64 ID string `json:"id"`
65 AppID string `json:"app_id"`
66 Epoch uint64 `json:"epoch"`
67 TombstoneETag string `json:"tombstone_etag"`
68 DeletedAt int64 `json:"deleted_at"`
69 State string `json:"state"`
70}
71type AppAdmission struct {
72 V int `json:"v"`
73 AppID string `json:"app_id"`
74 Revision uint64 `json:"revision"`
75 Epoch uint64 `json:"epoch"`
76 State string `json:"state"`
77 Op *AppOperation `json:"op"`
78 Registrations map[string]Registration `json:"registrations"`
79 Effects map[string]Effect `json:"effects"`
80 LastPurge *PurgeMark `json:"last_purge"`
81 LastRestore *RestoreMark `json:"last_restore,omitempty"`
82}
83type AdmissionObserved struct {
84 Key string
85 ETag string
86 Value AppAdmission
87}
88
89// Kinds accepted for an exclusive operation. "purging" is reachable only
90// through BeginPurge, which consumes a reserved journal.
91const (
92 OpDeleting = "deleting"
93 OpPlacing = "placing"
94 OpPurging = "purging"
95 OpAppMutation = "app-mutation"
96)
97
98func AdmissionKey(account, app string) (string, error) {
99 if !controlname.ValidAccount(account) || !controlname.ValidApp(app) {
100 return "", ErrIdentity
101 }
102 return "apps/" + account + "/" + app + "/admission.json", nil
103}
104
105// ReadAdmission is a read. It works through the data boundary; it never
106// initializes, repairs or interprets absence.
107func ReadAdmission(ctx context.Context, s release.Store, account, app, appID string) (AdmissionObserved, error) {
108 if appID == "" {
109 return AdmissionObserved{}, ErrIdentity
110 }
111 key, err := AdmissionKey(account, app)
112 if err != nil {
113 return AdmissionObserved{}, err
114 }
115 body, etag, err := s.Get(ctx, key)
116 if err != nil {
117 return AdmissionObserved{}, err
118 }
119 var a AppAdmission
120 if err = decode(body, &a); err != nil {
121 return AdmissionObserved{}, err
122 }
123 if !validAdmission(a) || etag == "" || a.AppID != appID {
124 return AdmissionObserved{}, ErrIdentity
125 }
126 return AdmissionObserved{Key: key, ETag: etag, Value: a}, nil
127}
128
129// ReadCanonicalAdmission reads the object's own app_id and then validates it.
130// A missing object is os.ErrNotExist; a blank or invalid one is ErrIdentity.
131func ReadCanonicalAdmission(ctx context.Context, s release.Store, account, app string) (AdmissionObserved, error) {
132 key, err := AdmissionKey(account, app)
133 if err != nil {
134 return AdmissionObserved{}, err
135 }
136 body, _, err := s.Get(ctx, key)
137 if err != nil {
138 return AdmissionObserved{}, err
139 }
140 var identity struct {
141 AppID string `json:"app_id"`
142 }
143 if err = json.Unmarshal(body, &identity); err != nil {
144 return AdmissionObserved{}, err
145 }
146 if identity.AppID == "" {
147 return AdmissionObserved{}, ErrIdentity
148 }
149 return ReadAdmission(ctx, s, account, app, identity.AppID)
150}
151
152func validAdmission(a AppAdmission) bool {
153 for id, r := range a.Registrations {
154 if r.Kind != RegistrationIngest {
155 continue
156 }
157 if id == "" || r.Host != "" || r.Claim != (SavedClaim{}) || r.AppID != a.AppID || r.Epoch != a.Epoch || !validRegistrationOwner(r.Owner) || r.Owner.Console == "" || r.Expires.IsZero() || r.Incarnation != "" || !r.CreatedAt.IsZero() {
158 return false
159 }
160 }
161 return a.V == 1 && a.AppID != "" && a.Revision != 0 && a.Epoch != 0 &&
162 (a.State == "open" || a.State == "closed") && a.Registrations != nil && a.Effects != nil &&
163 (a.Op == nil || (a.Op.ID != "" && a.Op.Kind != "" && a.Op.AppID == a.AppID))
164}
165
166// CompletedAdmission is the retired-incarnation predicate shared by
167// enrollment, materialization and the purge-request discard.
168func CompletedAdmission(a AppAdmission) bool {
169 if a.State != "closed" || a.Op != nil || len(a.Registrations) != 0 || a.LastPurge == nil || a.LastPurge.AppID != a.AppID || a.LastPurge.ID == "" {
170 return false
171 }
172 for _, effect := range a.Effects {
173 if effect.State != "complete" {
174 return false
175 }
176 }
177 return true
178}
179
180func validRegistrationOwner(o RegistrationOwner) bool {
181 return (o.Console != "") != (o.Box != "") && o.Enrollment != "" && o.Controller != ""
182}
183func RegistrationLive(r Registration, now time.Time) bool {
184 return now.Before(r.Expires.Add(-RegistrationSkew))
185}
186func RegistrationDrained(r Registration, now time.Time) bool {
187 return !now.Before(r.Expires.Add(RegistrationSkew))
188}
189func SavedObservation(o hostclaim.Observed) SavedClaim {
190 if o.Absent {
191 return SavedClaim{Absent: true}
192 }
193 return SavedClaim{ETag: o.ETag, Gen: o.Claim.Gen}
194}
195
196// ---- pure transitions (unexported; the ControlWriter is the only caller) ----
197
198func registerStart(a *AppAdmission, id string, r Registration, now time.Time) error {
199 if id == "" || a.State != "open" || a.Op != nil || r.AppID != a.AppID || r.Epoch != a.Epoch || !validRegistrationOwner(r.Owner) || r.Kind == "" || now.IsZero() {
200 return ErrPending
201 }
202 if r.Kind == registrationPurge || r.Kind == RegistrationIngest {
203 // Ingest and purge reservations have no host or saved claim.
204 if r.Kind == RegistrationIngest && (r.Owner.Console == "" || r.Incarnation != "" || !r.CreatedAt.IsZero()) {
205 return ErrIdentity
206 }
207 if r.Host != "" || r.Claim != (SavedClaim{}) {
208 return ErrIdentity
209 }
210 } else {
211 if !controlname.ValidHostShape(r.Host) {
212 return ErrPending
213 }
214 if (r.Claim.Absent && (r.Claim.ETag != "" || r.Claim.Gen != 0)) || (!r.Claim.Absent && (r.Claim.ETag == "" || r.Claim.Gen == 0)) {
215 return ErrIdentity
216 }
217 }
218 if _, ok := a.Registrations[id]; ok {
219 return ErrIdentity
220 }
221 r.Expires = now.UTC().Add(RegistrationTTL)
222 a.Registrations[id] = r
223 return nil
224}
225
226// renewStart deliberately works after admission closed: an accepted
227// registration may keep renewing until its own operation settles.
228func renewStart(a *AppAdmission, id string, owner RegistrationOwner, now time.Time) error {
229 r, ok := a.Registrations[id]
230 if !ok || r.Owner != owner || r.AppID != a.AppID || r.Epoch != a.Epoch || !RegistrationLive(r, now) {
231 return ErrPending
232 }
233 r.Expires = now.UTC().Add(RegistrationTTL)
234 a.Registrations[id] = r
235 return nil
236}
237
238func bindStart(a *AppAdmission, id string, owner RegistrationOwner, created time.Time, incarnation string, now time.Time) error {
239 r, ok := a.Registrations[id]
240 if !ok || r.Owner != owner || r.AppID != a.AppID || r.Epoch != a.Epoch || !RegistrationLive(r, now) || created.IsZero() || incarnation == "" {
241 return ErrPending
242 }
243 if r.Incarnation != "" && (r.Incarnation != incarnation || !r.CreatedAt.Equal(created)) {
244 return ErrIdentity
245 }
246 r.CreatedAt = created.UTC()
247 r.Incarnation = incarnation
248 a.Registrations[id] = r
249 return nil
250}
251
252func exclusiveAdmission(a *AppAdmission, op AppOperation) error {
253 if op.ID == "" || op.Kind == "" || op.AppID != a.AppID {
254 return ErrIdentity
255 }
256 if a.Op != nil {
257 if a.State == "closed" && reflect.DeepEqual(*a.Op, op) {
258 return nil
259 }
260 return ErrPending
261 }
262 // A completed (closed, op-less) admission never reopens here; that is
263 // the dedicated recreation transition in EnrollNewApp.
264 if a.State != "open" {
265 return ErrPending
266 }
267 if op.Intent == nil {
268 op.Intent = map[string]string{}
269 }
270 a.State = "closed"
271 a.Op = &op
272 return nil
273}
274
275func prepareEffect(a *AppAdmission, e Effect) error {
276 if e.ID == "" || e.OperationID == "" || e.AppID != a.AppID || e.Resource == "" || e.Action == "" || e.State != "prepared" || !validRegistrationOwner(e.Owner) {
277 return ErrIdentity
278 }
279 if old, ok := a.Effects[e.ID]; ok {
280 if reflect.DeepEqual(old, e) {
281 return nil
282 }
283 return ErrIdentity
284 }
285 for _, old := range a.Effects {
286 if old.Resource == e.Resource && old.State != "complete" {
287 return ErrPending
288 }
289 }
290 a.Effects[e.ID] = e
291 return nil
292}
293
294func advanceEffect(a *AppAdmission, id string, owner RegistrationOwner, state, result string) error {
295 e, ok := a.Effects[id]
296 if !ok || e.Owner != owner || e.AppID != a.AppID {
297 return ErrIdentity
298 }
299 if e.State == state && e.ResultReference == result {
300 return nil
301 }
302 allowed := e.State == "prepared" && state == "issued" ||
303 e.State == "issued" && (state == "uncertain" || state == "complete") ||
304 e.State == "uncertain" && state == "complete"
305 if !allowed {
306 return ErrPending
307 }
308 if state == "complete" && result == "" {
309 return ErrIdentity
310 }
311 e.State = state
312 e.ResultReference = result
313 a.Effects[id] = e
314 return nil
315}
316
317func effectsSettled(a *AppAdmission, operation string) bool {
318 for _, e := range a.Effects {
319 if (operation == "" || e.OperationID == operation) && e.State != "complete" {
320 return false
321 }
322 }
323 return true
324}
325
326// nextAdmission copies before through JSON so a refused mutation never
327// aliases the caller's maps, and enforces the invariants every transition
328// shares: same version and app_id; epoch changes only through reopen.
329func nextAdmission(before AdmissionObserved, mutate func(*AppAdmission) error) (AppAdmission, error) {
330 if before.ETag == "" || before.Key == "" || before.Value.Revision == math.MaxUint64 {
331 return AppAdmission{}, ErrIdentity
332 }
333 body, err := json.Marshal(before.Value)
334 if err != nil {
335 return AppAdmission{}, err
336 }
337 var next AppAdmission
338 if err = json.Unmarshal(body, &next); err != nil {
339 return AppAdmission{}, err
340 }
341 if err = mutate(&next); err != nil {
342 return AppAdmission{}, err
343 }
344 reopen := before.Value.State == "closed" && next.State == "open" && before.Value.Op != nil && next.Op == nil &&
345 before.Value.Epoch < math.MaxUint64 && next.Epoch == before.Value.Epoch+1 && len(next.Registrations) == 0
346 if next.AppID != before.Value.AppID || (!reopen && next.Epoch != before.Value.Epoch) || next.V != 1 || !validAdmission(withRevision(next, 1)) {
347 return AppAdmission{}, ErrIdentity
348 }
349 // last_purge is written only by CompletePurge and carried by recreation.
350 if !reflect.DeepEqual(next.LastPurge, before.Value.LastPurge) && !purgeCompletion(before.Value, next) {
351 return AppAdmission{}, ErrIdentity
352 }
353 next.Revision = before.Value.Revision + 1
354 return next, nil
355}
356
357func withRevision(a AppAdmission, r uint64) AppAdmission { a.Revision = r; return a }
358
359func purgeCompletion(before, next AppAdmission) bool {
360 return before.Op != nil && before.Op.Kind == OpPurging && next.Op == nil && next.State == "closed" &&
361 next.LastPurge != nil && next.LastPurge.ID == before.Op.ID && next.LastPurge.AppID == before.AppID
362}
363