amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "bytes"
5 "context"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "log"
10 "os"
11 "path/filepath"
12 "reflect"
13 "slices"
14 "strings"
15 "sync"
16 "testing"
17 "time"
18
19 "github.com/carlosframework/platform/internal/activator"
20 "github.com/carlosframework/platform/internal/appconfig"
21 "github.com/carlosframework/platform/internal/cgroup"
22 "github.com/carlosframework/platform/internal/hostwrite"
23 "github.com/carlosframework/platform/internal/instancectl"
24 "github.com/carlosframework/platform/internal/registry"
25 "github.com/carlosframework/platform/internal/release"
26 "github.com/carlosframework/platform/internal/rollout"
27)
28
29// rolloutFixture is an activator over a fresh registry, built with the
30// generations switch as gens says (runAgent sets it from swapArming before
31// New). The package's rollout state is reset when the test ends.
32func rolloutFixture(t *testing.T, gens bool) (*activator.Service, *registry.Registry, string) {
33 t.Helper()
34 return rolloutFixtureWith(t, gens, &activator.SplitBackend{})
35}
36
37// rolloutFixtureWith is rolloutFixture over backend.
38func rolloutFixtureWith(t *testing.T, gens bool, backend activator.Backend) (*activator.Service, *registry.Registry, string) {
39 t.Helper()
40 prev := activator.GenerationsEnabled()
41 activator.SetGenerations(gens)
42 t.Cleanup(func() { activator.SetGenerations(prev) })
43 resetRollout(t)
44 dd := t.TempDir()
45 reg, err := registry.Open(filepath.Join(dd, "registry.db"))
46 if err != nil {
47 t.Fatal(err)
48 }
49 t.Cleanup(func() { reg.Close() })
50 svc := activator.New(activator.Config{Bucket: "b", Idle: time.Hour}, backend, reg)
51 return svc, reg, dd
52}
53
54func resetRollout(t *testing.T) {
55 t.Helper()
56 clear := func() {
57 rolloutMgr.Store(nil)
58 rolloutDown.Store(nil)
59 rolloutDeclined.Store(nil)
60 declinedHolds.Store(nil)
61 select {
62 case <-statusNudge:
63 default:
64 }
65 }
66 clear()
67 t.Cleanup(clear)
68}
69
70func overlapEnv(pub string) func(string) string {
71 env := map[string]string{"CARLOS_SWAP": "overlap", "CARLOS_RELEASE_PUBKEY": pub}
72 return func(k string) string { return env[k] }
73}
74
75func noEnv(string) string { return "" }
76
77// stubPlacementCheck replaces swapArming's arm-time placement check with
78// one that answers err, and counts its calls. A temp-dir cgroup root can
79// never pass the real check (cgroup.Manager.VerifyPlacement), so every test
80// that expects swapArming to arm uses this with a nil err.
81func stubPlacementCheck(t *testing.T, err error) *int {
82 t.Helper()
83 orig := placementCheck
84 calls := new(int)
85 placementCheck = func(*cgroup.Manager) error { *calls++; return err }
86 t.Cleanup(func() { placementCheck = orig })
87 return calls
88}
89
90// forbidPlacementCheck fails the test if swapArming runs the placement
91// check: with the flag off, or after an earlier check declined, it must not.
92func forbidPlacementCheck(t *testing.T) {
93 t.Helper()
94 orig := placementCheck
95 placementCheck = func(*cgroup.Manager) error {
96 t.Error("swapArming ran the placement check")
97 return nil
98 }
99 t.Cleanup(func() { placementCheck = orig })
100}
101
102// armableBackend is an exec backend that passes every check swapArming makes
103// before the placement check: claims off, a delegated subtree (over a temp
104// root), and a control channel that speaks the generation verbs.
105func armableBackend(t *testing.T) *activator.ExecBackend {
106 t.Helper()
107 root := t.TempDir()
108 for _, f := range []string{"cgroup.procs", "cgroup.subtree_control"} {
109 if err := os.WriteFile(filepath.Join(root, f), nil, 0o644); err != nil {
110 t.Fatal(err)
111 }
112 }
113 cgm, err := cgroup.SetupAt(root, t.Logf)
114 if err != nil {
115 t.Fatal(err)
116 }
117 return &activator.ExecBackend{Cgroups: cgm, Control: instancectl.NewClient(filepath.Join(t.TempDir(), "ctl.sock"))}
118}
119
120func unusedStore(context.Context) (release.Store, error) { return nil, errors.New("unused") }
121
122func funcPtr(f any) uintptr { return reflect.ValueOf(f).Pointer() }
123
124// legsAreTodays asserts every leg the scheduler could take over is today's:
125// the converge leg, the adopt bounce, config sync's callbacks and the
126// restart leg's bounce.
127func legsAreTodays(t *testing.T, svc *activator.Service, reg *registry.Registry) {
128 t.Helper()
129 if fastConvergeLeg(svc, reg) == nil {
130 t.Error("no converge leg: convergeAdopted must run")
131 }
132 if b := adoptBounceFn(svc); b == nil || funcPtr(b) != funcPtr(svc.BounceAdopted) {
133 t.Error("the adopt bounce is not BounceAdopted")
134 }
135 if _, ok := rolloutConfigCallbacks(context.Background(), release.NewDirStore(t.TempDir()), configLegs{}); ok {
136 t.Error("config sync left today's SyncExec")
137 }
138 ctx, s := genKit(t)
139 routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")}
140 var bounced []string
141 bounce := func(h string) bool { bounced = append(bounced, h); return true }
142 mem := newRestartMemory()
143 askRestart(t, ctx, s, "bab", "eleven")
144 restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
145 askRestart(t, ctx, s, "bab", "eleven")
146 restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
147 if !slices.Equal(bounced, []string{"exec.example.com"}) {
148 t.Errorf("restart bounced %v, want today's bounce of the exec instance", bounced)
149 }
150}
151
152// Cross-plan review focus 1: the flag off changes nothing.
153func TestTheFlagOffArmsNothing(t *testing.T) {
154 svc, reg, dd := rolloutFixture(t, false)
155 opened := false
156 m := armRollout(context.Background(), noEnv, svc, reg, dd, nil,
157 func(context.Context) (release.Store, error) { opened = true; return nil, errors.New("unused") })
158 if m != nil || armedRollout() != nil || opened {
159 t.Fatal("CARLOS_SWAP unset armed the scheduler")
160 }
161 if svc.Planner != nil || svc.OnGen != nil || svc.OnDrift != nil || svc.CoalesceRecycle != nil || svc.GenerationsFile != "" {
162 t.Fatal("CARLOS_SWAP unset set an activator hook")
163 }
164 if adoptBounceFn(svc) == nil {
165 t.Fatal("the adopt bounce must stay BounceAdopted")
166 }
167 forbidPlacementCheck(t)
168 if swapArming(noEnv, nil) || activator.GenerationsEnabled() {
169 t.Fatal("CARLOS_SWAP unset turned generations on")
170 }
171 if r := rolloutStatusProbe(); r != nil {
172 t.Fatalf("status with the flag off = %+v, want nil", r)
173 }
174}
175
176// Carried obligation 5: with the flag off every leg is today's, and the only
177// hook set is armAlways's OnWorkerServing.
178func TestTheFlagOffLegsAreTodays(t *testing.T) {
179 svc, reg, dd := rolloutFixture(t, false)
180 armAlways(svc, reg, dd)
181 armRollout(context.Background(), noEnv, svc, reg, dd, nil, unusedStore)
182 legsAreTodays(t, svc, reg)
183 if len(statusNudge) != 0 {
184 t.Fatal("the status loop was nudged with the flag off")
185 }
186}
187
188// PF14 (plan 4 R17): runAgent sets the generations file whatever the flag,
189// so a worker's captured serving build is pinned with the flag off too.
190// armAlways publishes the worker's serving version and touches nothing
191// else; armRollout never sets the file either.
192func TestTheFlagOffStillPublishesWorkerArtifacts(t *testing.T) {
193 svc, reg, dd := rolloutFixture(t, false)
194 armAlways(svc, reg, dd)
195 armRollout(context.Background(), noEnv, svc, reg, dd, nil, unusedStore)
196 if svc.OnWorkerServing == nil {
197 t.Fatal("no worker serving publication with the flag off")
198 }
199 if svc.GenerationsFile != "" {
200 t.Fatalf("generations file = %q: armAlways and armRollout must leave it to runAgent", svc.GenerationsFile)
201 }
202}
203
204func TestTheFlagOnArmsTheScheduler(t *testing.T) {
205 svc, reg, dd := rolloutFixture(t, true)
206 svc.GenerationsFile = release.GenerationsPath(dd) // as runAgent sets it, before armRollout
207 _, pub, err := release.GenerateKey()
208 if err != nil {
209 t.Fatal(err)
210 }
211 if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready",
212 Hibernate: true, Version: "v2"}); err != nil {
213 t.Fatal(err)
214 }
215 store := release.NewDirStore(t.TempDir())
216 ctx, cancel := context.WithCancel(context.Background()) // ends the schedulers' goroutines
217 t.Cleanup(cancel)
218 m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil,
219 func(context.Context) (release.Store, error) { return store, nil })
220 if m == nil || armedRollout() != m {
221 t.Fatal("CARLOS_SWAP=overlap did not arm the scheduler")
222 }
223 if svc.Planner == nil || svc.OnGen == nil || svc.OnDrift == nil || svc.CoalesceRecycle == nil {
224 t.Fatal("a hook is missing")
225 }
226 if svc.GenerationsFile != release.GenerationsPath(dd) {
227 t.Fatalf("generations file = %q", svc.GenerationsFile)
228 }
229 // The legs are the scheduler's.
230 if b := adoptBounceFn(svc); b == nil || funcPtr(b) != funcPtr(m.Adopted) {
231 t.Fatal("the adopt bounce is not the scheduler's Adopted")
232 }
233 if fastConvergeLeg(svc, reg) == nil {
234 t.Fatal("no converge leg: the scheduler's pass must run")
235 }
236 cb, ok := rolloutConfigCallbacks(context.Background(), store, configLegs{})
237 if !ok || cb.Delivered == nil || cb.Ack == nil || cb.Restart == nil || cb.Restart("a.example.com") != nil {
238 t.Fatalf("config callbacks = %+v, %v: want the scheduler's delivery and deferred ack, and no restart", cb, ok)
239 }
240 // The pass at arming saw the managed route, and the status probe is
241 // the scheduler's report.
242 r := rolloutStatusProbe()
243 if r == nil || r.Unavailable != "" {
244 t.Fatalf("status = %+v, want the scheduler's report", r)
245 }
246 if got := r.Hosts["a.example.com"]; got.Installed != "v2" {
247 t.Fatalf("status host = %+v, want the managed route installed on v2", got)
248 }
249}
250
251// Spec §6.4, §8.3, §8.4, P2 #1: with the flag on, generations on and the
252// scheduler's state unavailable, managed starts are refused, the reason
253// code reported and the bounce legs stand down. Falling back to today's
254// live-link wakes and legacy convergence (a nil Planner) would exec the
255// failed installed build the unreadable records hold down.
256func TestUnavailableStateRefusesManagedStarts(t *testing.T) {
257 _, pub, err := release.GenerateKey()
258 if err != nil {
259 t.Fatal(err)
260 }
261 dirStore := func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }
262 for name, tc := range map[string]struct {
263 env func(string) string
264 store func(context.Context) (release.Store, error)
265 book bool // an unreadable failure book in the data dir
266 code string
267 }{
268 "no deploy store": {overlapEnv(pub),
269 func(context.Context) (release.Store, error) { return nil, errors.New("no deployment store") }, false, rolloutNoStore},
270 "no public key": {overlapEnv(""), dirStore, false, rolloutNoReleaseKey},
271 "unreadable book": {overlapEnv(pub), dirStore, true, rolloutBookUnreadable},
272 "book with a bad schema": {overlapEnv(pub), dirStore, true, rolloutBookUnreadable},
273 } {
274 t.Run(name, func(t *testing.T) {
275 svc, reg, dd := rolloutFixture(t, true)
276 if tc.book {
277 body := "{not json"
278 if name == "book with a bad schema" {
279 body = `{"schema": 9}`
280 }
281 if err := os.WriteFile(filepath.Join(dd, rollout.FailureFile), []byte(body), 0o644); err != nil {
282 t.Fatal(err)
283 }
284 }
285 m := armRollout(context.Background(), tc.env, svc, reg, dd, nil, tc.store)
286 if m != nil || armedRollout() != nil {
287 t.Fatal("armed without its required state")
288 }
289 if svc.Planner == nil {
290 t.Fatal("no Planner: the box fell back to today's live-link wakes")
291 }
292 managed := registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true, Version: "v2"}
293 p := svc.Planner(context.Background(), managed)
294 if p.Refuse == "" {
295 t.Fatalf("managed wake planned %+v, want refused", p)
296 }
297 if p.Refuse != refuseTexts[tc.code] {
298 t.Fatalf("refusal %q, want the fixed text for %s (never the raw error)", p.Refuse, tc.code)
299 }
300 unit := registry.Route{Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"}
301 if p := svc.Planner(context.Background(), unit); p.Refuse != "" || p.Boot.StagedPath != "" {
302 t.Fatalf("an unmanaged wake was changed: %+v", p)
303 }
304 if r := rolloutStatusProbe(); r == nil || r.Unavailable != tc.code {
305 t.Fatalf("status = %+v, want the reason code %q", r, tc.code)
306 }
307 // The bounce legs stand down.
308 if adoptBounceFn(svc) != nil {
309 t.Fatal("the adopt bounce still stops instances whose next wake is refused")
310 }
311 if fastConvergeLeg(svc, reg) != nil {
312 t.Fatal("legacy convergence still stops instances whose next wake is refused")
313 }
314 if svc.OnDrift == nil {
315 t.Fatal("the sweep's convergence stop is still armed")
316 }
317 if svc.CoalesceRecycle == nil || svc.CoalesceRecycle("bot.internal") {
318 t.Fatal("a worker recycle can still converge onto the live link")
319 }
320 if svc.OnGen != nil {
321 t.Fatal("OnGen set with no scheduler to hear it")
322 }
323 store := release.NewDirStore(t.TempDir())
324 cb, ok := rolloutConfigCallbacks(context.Background(), store, configLegs{})
325 // Nothing is awake here: the restart is none, and says the file
326 // is read on the next wake (final review F7 (3)).
327 if !ok || cb.Delivered != nil || cb.Restart == nil || !errors.Is(cb.Restart("a.example.com"), appconfig.ErrDeliveredOnWake) || cb.Ack == nil {
328 t.Fatalf("config callbacks = %+v, %v: want delivery and the ack as today, and no restart", cb, ok)
329 }
330 ctx, s := genKit(t)
331 routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")}
332 var bounced []string
333 bounce := func(h string) bool { bounced = append(bounced, h); return true }
334 mem := newRestartMemory()
335 askRestart(t, ctx, s, "bab", "eleven")
336 restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem)
337 askRestart(t, ctx, s, "bab", "eleven")
338 if !restartRequestLeg(ctx, routes, s, t.TempDir(), bounce, nil, mem) {
339 t.Fatal("the root half of the restart was not asked for")
340 }
341 if len(bounced) != 0 {
342 t.Fatalf("restart bounced %v while managed starts are refused", bounced)
343 }
344 })
345 }
346}
347
348// leaveIntent leaves in's intent for host in dd's failure book, as an agent
349// that died mid-execution would have.
350func leaveIntent(t *testing.T, dd, host string, in rollout.Intent) {
351 t.Helper()
352 b, err := rollout.OpenFailureBook(dd)
353 if err != nil {
354 t.Fatal(err)
355 }
356 if err := b.Begin(host, in); err != nil {
357 t.Fatal(err)
358 }
359}
360
361// Task 5 rule 5: an armed box whose book holds an intent an earlier agent
362// left turns it into the failure it may have been, and saves that before
363// the scheduler runs: a reopen finds the record and no intent.
364func TestAnArmedBoxReconcilesAnIntentLeftInItsBook(t *testing.T) {
365 svc, reg, dd := rolloutFixture(t, true)
366 svc.GenerationsFile = release.GenerationsPath(dd)
367 _, pub, err := release.GenerateKey()
368 if err != nil {
369 t.Fatal(err)
370 }
371 if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready",
372 Hibernate: true, Version: "v2", GoodVersion: "v1", GoodDigest: "sha256:v1"}); err != nil {
373 t.Fatal(err)
374 }
375 leaveIntent(t, dd, "a.example.com", rollout.Intent{Account: "bab", App: "blog", Build: "v2", Mode: "hold", ID: "a3"})
376 store := release.NewDirStore(t.TempDir())
377 ctx, cancel := context.WithCancel(context.Background())
378 t.Cleanup(cancel)
379 m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, func(context.Context) (release.Store, error) { return store, nil })
380 if m == nil || rolloutUnavailable() != "" {
381 t.Fatalf("not armed (%q)", rolloutUnavailable())
382 }
383 b, err := rollout.OpenFailureBook(dd)
384 if err != nil {
385 t.Fatal(err)
386 }
387 if b.Reconciled() != 0 {
388 t.Fatal("the reconciled book was not saved: the intent is still on disk")
389 }
390 if !b.HeldDown("bab", "blog", "v2", "a.example.com") {
391 t.Fatal("the hold intent's failure was not kept")
392 }
393 if p := svc.Planner(ctx, *mustGet(t, reg, "a.example.com")); p.Refuse != rollout.HeldDownRefusal("v2") {
394 t.Fatalf("plan %+v: want the held-down refusal", p)
395 }
396}
397
398// Task 5 rule 5: a reconciled book the box cannot save refuses every
399// managed start, as an unreadable one does: the scheduler must not run while
400// a restart would find the intents again with no way to keep their records.
401func TestAnArmedBoxThatCannotSaveItsReconciledBookRefusesManagedStarts(t *testing.T) {
402 if os.Geteuid() == 0 {
403 t.Skip("running as root: a read-only directory does not fail a save")
404 }
405 svc, reg, dd := rolloutFixture(t, true)
406 _, pub, err := release.GenerateKey()
407 if err != nil {
408 t.Fatal(err)
409 }
410 leaveIntent(t, dd, "a.example.com", rollout.Intent{Account: "bab", App: "blog", Build: "v2", Mode: "overlap", ID: "w4"})
411 if err := os.Chmod(dd, 0o500); err != nil {
412 t.Fatal(err)
413 }
414 t.Cleanup(func() { os.Chmod(dd, 0o700) })
415 m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil,
416 func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil })
417 if m != nil || armedRollout() != nil {
418 t.Fatal("armed over a book it could not save")
419 }
420 if rolloutUnavailable() != rolloutBookUnreadable {
421 t.Fatalf("unavailable %q, want %q", rolloutUnavailable(), rolloutBookUnreadable)
422 }
423 managed := registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true, Version: "v2"}
424 if p := svc.Planner(context.Background(), managed); p.Refuse != refuseTexts[rolloutBookUnreadable] {
425 t.Fatalf("plan %+v: want the unreadable-book refusal", p)
426 }
427}
428
429// Final review round 1, finding 3, restated for the hard rule (readiness
430// round 2 Task 1; FailureBook's doc, rules 2 to 5): every open of the book
431// owes a save before the scheduler arms, intents or none. A new book starts
432// at sequence 0 with a crypto/rand identity and is written as schema 3 with
433// durable_epochs; a box that cannot save that could publish epochs it never
434// stored, so it refuses managed starts. A box that can save it arms, with
435// the identity and the durable marker on disk.
436func TestAnArmedBoxSavesItsBookOnOpenOrRefusesManagedStarts(t *testing.T) {
437 if os.Geteuid() == 0 {
438 t.Skip("running as root: a read-only directory does not fail a save")
439 }
440 _, pub, err := release.GenerateKey()
441 if err != nil {
442 t.Fatal(err)
443 }
444 store := func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }
445
446 svc, reg, dd := rolloutFixture(t, true)
447 if err := os.Chmod(dd, 0o500); err != nil {
448 t.Fatal(err)
449 }
450 locked := dd
451 t.Cleanup(func() { os.Chmod(locked, 0o700) })
452 if m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, store); m != nil || armedRollout() != nil {
453 t.Fatal("armed over a book whose open it could not save")
454 }
455 if rolloutUnavailable() != rolloutBookUnreadable {
456 t.Fatalf("unavailable %q, want %q", rolloutUnavailable(), rolloutBookUnreadable)
457 }
458
459 svc, reg, dd = rolloutFixture(t, true)
460 if m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, store); m == nil {
461 t.Fatal("did not arm over a writable data dir")
462 }
463 raw, err := os.ReadFile(filepath.Join(dd, rollout.FailureFile))
464 if err != nil {
465 t.Fatalf("the open's advance was not saved before arming: %v", err)
466 }
467 var f struct {
468 Schema int `json:"schema"`
469 Seq uint64 `json:"seq"`
470 Book string `json:"book"`
471 Durable bool `json:"durable_epochs"`
472 Records []json.RawMessage `json:"records"`
473 }
474 if err := json.Unmarshal(raw, &f); err != nil {
475 t.Fatalf("saved book %s: %v", raw, err)
476 }
477 if f.Schema != 3 || !f.Durable {
478 t.Fatalf("saved book %s: want schema 3 with durable_epochs true", raw)
479 }
480 if id, err := hex.DecodeString(f.Book); err != nil || len(id) != 16 {
481 t.Fatalf("saved book %s (%v): want a 16-byte hex identity", raw, err)
482 }
483 // This fixture has no records, intents or markers, so the highest value
484 // the book ever held is 0 and a new book hands out none before its first
485 // change: the saved sequence is at least that, and no more than the
486 // open itself needs (the old 2^20 advance is gone).
487 if len(f.Records) != 0 || f.Seq >= 1<<20 {
488 t.Fatalf("saved book %s: want an empty new book with a small sequence", raw)
489 }
490}
491
492// The unreadable-book refusal is one approved text, shared with the
493// scheduler's refusal of a wake whose intent it could not write (Task 5).
494func TestTheUnreadableBookRefusalIsShared(t *testing.T) {
495 if refuseTexts[rolloutBookUnreadable] != rollout.BookUnreadableRefusal {
496 t.Fatalf("host %q, rollout %q", refuseTexts[rolloutBookUnreadable], rollout.BookUnreadableRefusal)
497 }
498}
499
500func mustGet(t *testing.T, reg *registry.Registry, host string) *registry.Route {
501 t.Helper()
502 rt, err := reg.Get(host)
503 if err != nil || rt == nil {
504 t.Fatalf("get %s: %v", host, err)
505 }
506 return rt
507}
508
509// PF15: with the flag on but no delegated cgroup subtree, the flag is
510// refused before the activator exists. Generations and containment stay
511// off, no manager is armed, no hook is set, every leg is today's, and the
512// status carries the reason code.
513func TestNoDelegationDeclinesTheFlag(t *testing.T) {
514 _, pub, err := release.GenerateKey()
515 if err != nil {
516 t.Fatal(err)
517 }
518 resetRollout(t)
519 forbidPlacementCheck(t)
520 eb := &activator.ExecBackend{Control: instancectl.NewClient(filepath.Join(t.TempDir(), "ctl.sock"))}
521 if swapArming(overlapEnv(pub), eb) {
522 t.Fatal("armed generations with no delegated cgroup subtree")
523 }
524 if got := rolloutUnarmed(); got != rolloutNoDelegation {
525 t.Fatalf("declined with %q, want %q", got, rolloutNoDelegation)
526 }
527 // runAgent: SetGenerations(swapOn), then New, then armRollout.
528 svc, reg, dd := rolloutFixture(t, false)
529 declineRollout(rolloutNoDelegation) // rolloutFixture reset it; runAgent would not
530 opened := false
531 m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil,
532 func(context.Context) (release.Store, error) {
533 opened = true
534 return release.NewDirStore(t.TempDir()), nil
535 })
536 if m != nil || armedRollout() != nil || opened {
537 t.Fatal("armed a scheduler over an activator without generations")
538 }
539 if svc.Planner != nil || svc.OnGen != nil || svc.OnDrift != nil || svc.CoalesceRecycle != nil {
540 t.Fatal("a declined flag set an activator hook")
541 }
542 if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutNoDelegation {
543 t.Fatalf("status = %+v, want the reason code %q", r, rolloutNoDelegation)
544 }
545 if rolloutUnavailable() != "" {
546 t.Fatal("a declined flag refuses managed starts")
547 }
548 legsAreTodays(t, svc, reg)
549}
550
551// The armed half of PF15: a delegated subtree and a control channel that
552// speaks the generation verbs arm generations; a channel without them does
553// not.
554func TestDelegationArmsTheFlag(t *testing.T) {
555 _, pub, err := release.GenerateKey()
556 if err != nil {
557 t.Fatal(err)
558 }
559 resetRollout(t)
560 calls := stubPlacementCheck(t, nil)
561 eb := armableBackend(t)
562 if !swapArming(overlapEnv(pub), eb) || rolloutUnarmed() != "" {
563 t.Fatalf("delegation present: declined with %q", rolloutUnarmed())
564 }
565 if *calls != 1 {
566 t.Fatalf("the armed flag ran the placement check %d times, want once", *calls)
567 }
568 if swapArming(noEnv, eb) {
569 t.Fatal("the flag off armed generations")
570 }
571 if *calls != 1 {
572 t.Fatal("the flag off ran the placement check")
573 }
574 eb.Control = nil
575 if swapArming(overlapEnv(pub), eb) || rolloutUnarmed() != rolloutNoGenControl {
576 t.Fatalf("no generation control: armed, or declined with %q", rolloutUnarmed())
577 }
578 if *calls != 1 {
579 t.Fatal("a box that declined on its generation control still ran the placement check")
580 }
581}
582
583// The second dev-box check (#272): a box armed CARLOS_SWAP=overlap and then
584// failed every generation start, because no process could be placed in its
585// cgroup under carlos-edge.service's hardening. Arming now does one real
586// placement into a scratch cgroup and declines when it fails, with a code
587// of its own, a loud operator line, and the code in the box's status.
588func TestSwapArmingDeclinesWhenPlacementFails(t *testing.T) {
589 _, pub, err := release.GenerateKey()
590 if err != nil {
591 t.Fatal(err)
592 }
593 resetRollout(t)
594 var buf bytes.Buffer
595 log.SetOutput(&buf)
596 t.Cleanup(func() { log.SetOutput(os.Stderr) })
597 eb := armableBackend(t)
598 var got *cgroup.Manager
599 orig := placementCheck
600 placementCheck = func(m *cgroup.Manager) error { got = m; return errors.New("probe exited before exec") }
601 t.Cleanup(func() { placementCheck = orig })
602 if swapArming(overlapEnv(pub), eb) {
603 t.Fatal("armed with a failed placement check")
604 }
605 if got != eb.Cgroups {
606 t.Fatal("the placement check was not run against the backend's cgroup manager")
607 }
608 if rolloutUnarmed() != rolloutPlacementFailed {
609 t.Fatalf("declined as %q", rolloutUnarmed())
610 }
611 if rolloutUnavailable() != "" {
612 t.Fatal("a placement decline refuses managed starts")
613 }
614 if s := buf.String(); !strings.Contains(s, "REFUSING CARLOS_SWAP=overlap (cgroup-placement-failed)") ||
615 !strings.Contains(s, "probe exited before exec") {
616 t.Fatalf("no operator line naming the placement refusal and its error: %q", s)
617 }
618 if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutPlacementFailed {
619 t.Fatalf("status %+v: want Unavailable %q", r, rolloutPlacementFailed)
620 }
621}
622
623// A box that declined for a failed placement honours the failure records an
624// earlier armed run left, exactly as the other declined codes do
625// (honourDeclinedFailures keys on the activator built without generations,
626// not on the code), and publishes its own code beside the honoured book.
627func TestPlacementDeclineHonoursFailureRecords(t *testing.T) {
628 t.Setenv("CARLOS_OPT_DIR", t.TempDir())
629 svc, reg, dd := rolloutFixture(t, false)
630 acct := release.RouteAccount(managedRoute("x", ""))
631 writeFailureBook(t, dd, []rollout.FailureRecord{{
632 Account: acct, App: "blog", Build: "v2", Host: "sib.example.com", Cause: "adopt", Mode: "hold", Reason: "health",
633 }}, nil)
634 _, pub, err := release.GenerateKey()
635 if err != nil {
636 t.Fatal(err)
637 }
638 stubPlacementCheck(t, errors.New("the shim exited before placement"))
639 if swapArming(overlapEnv(pub), armableBackend(t)) {
640 t.Fatal("armed with a failed placement check")
641 }
642 m := armRollout(context.Background(), overlapEnv(pub), svc, reg, dd, nil, unusedStore)
643 if m != nil || armedRollout() != nil {
644 t.Fatal("a declined box armed the scheduler")
645 }
646 if heldHolds() == nil || svc.Planner == nil {
647 t.Fatal("a placement decline did not honour the failure records")
648 }
649 blocked := managedRoute("a.example.com", "v2")
650 blocked.GoodVersion, blocked.GoodDigest = "v1", "sha256:v1"
651 if p := svc.Planner(context.Background(), blocked); p.Refuse != refuseNoSwaps {
652 t.Fatalf("a wake onto a held build planned %+v, want refused", p)
653 }
654 if r := rolloutStatusProbe(); r == nil || r.Unavailable != rolloutPlacementFailed || r.FailureBook != rolloutBookHonoured {
655 t.Fatalf("status = %+v, want the decline code %q and the book %q", r, rolloutPlacementFailed, rolloutBookHonoured)
656 }
657}
658
659// A box with host claims on declines the swap flag (#272): claims-backed
660// launches cannot yet carry a swap's recovery choice, execution evidence,
661// generations file or good build. The claims check comes before delegation's,
662// so a box that has neither says claims.
663func TestSwapArmingDeclinesWhenHostClaimsAreOn(t *testing.T) {
664 resetRollout(t)
665 forbidPlacementCheck(t)
666 var buf bytes.Buffer
667 log.SetOutput(&buf)
668 t.Cleanup(func() { log.SetOutput(os.Stderr) })
669 // No cgroups either: the claims check must come first, so the code
670 // says claims, not delegation.
671 eb := &activator.ExecBackend{Cfg: activator.Config{HostWrites: hostwrite.Config{Enabled: true}}}
672 if swapArming(overlapEnv(""), eb) {
673 t.Fatal("armed a box with host claims on")
674 }
675 if got := rolloutUnarmed(); got != rolloutHostClaims {
676 t.Fatalf("decline code %q, want %q", got, rolloutHostClaims)
677 }
678 if !strings.Contains(buf.String(), "REFUSING CARLOS_SWAP=overlap (host-claims-on)") {
679 t.Fatalf("no operator line naming the claims refusal: %q", buf.String())
680 }
681 r := rolloutStatusProbe()
682 if r == nil || r.Unavailable != rolloutHostClaims {
683 t.Fatalf("status %+v: want Unavailable %q", r, rolloutHostClaims)
684 }
685}
686
687func TestSwapArmingIgnoresClaimsWithTheFlagOff(t *testing.T) {
688 resetRollout(t)
689 forbidPlacementCheck(t)
690 eb := &activator.ExecBackend{Cfg: activator.Config{HostWrites: hostwrite.Config{Enabled: true}}}
691 if swapArming(noEnv, eb) || rolloutUnarmed() != "" {
692 t.Fatal("flag off: a claims box must say nothing")
693 }
694}
695
696func TestRestartsFeedTheSchedulerWithTheirIdentity(t *testing.T) {
697 routes := []registry.Route{
698 {Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true},
699 {Host: "bot.internal", Kind: "sidecar", App: "blog", Account: "bab", Status: "ready"},
700 {Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"},
701 {Host: "o.example.com", Kind: "instance", App: "other", Account: "bab", Status: "ready", Hibernate: true},
702 }
703 k := [2]string{"bab", "blog"}
704 got := map[string]string{}
705 n := feedRestarts(routes, [][2]string{k}, nil, map[[2]string]string{k: "etag-7"}, func(h, id string) bool { got[h] = id; return true })
706 if n != 2 || got["a.example.com"] != "gen:etag-7" || got["bot.internal"] != "gen:etag-7" {
707 t.Fatalf("fed %d: %v; want the instance and the sidecar, never the named unit or another app", n, got)
708 }
709 scoped := []scopedRestart{{app: k, requests: []release.RestartRequest{
710 {ID: "r1", Scope: release.RestartScope{Kind: release.RestartScopeHost, Value: "a.example.com"}},
711 {ID: "r2", Scope: release.RestartScope{Kind: release.RestartScopeHost, Value: "a.example.com"}},
712 }}}
713 got = map[string]string{}
714 if n := feedRestarts(routes, nil, scoped, nil, func(h, id string) bool { got[h] = id; return true }); n != 1 || got["a.example.com"] != "scope:r2" {
715 t.Fatalf("fed %d: %v; want the narrowed host, identified by the newest request", n, got)
716 }
717}
718
719// Carried obligation 4 (T13): OnChange is a non-blocking nudge that
720// coalesces, so a burst of transitions is one early status tick.
721func TestOnChangeCoalescesIntoOneNudge(t *testing.T) {
722 resetRollout(t)
723 done := make(chan struct{})
724 go func() {
725 for range 100 {
726 nudgeStatus()
727 }
728 close(done)
729 }()
730 select {
731 case <-done:
732 case <-time.After(5 * time.Second):
733 t.Fatal("nudgeStatus blocked")
734 }
735 if len(statusNudge) != 1 {
736 t.Fatalf("%d nudges queued, want one", len(statusNudge))
737 }
738}
739
740// Carried obligation 3 (T11): OnWorkerServing returns at once even while
741// the registry write is slow, and the newest identity a host was handed is
742// the one that lands last.
743func TestTheWorkerServingWriteIsCheapAndKeepsTheNewest(t *testing.T) {
744 release1 := make(chan struct{})
745 var mu sync.Mutex
746 var writes []string
747 w := newServingWriter(func(host, ver string) error {
748 if ver == "v1" {
749 <-release1
750 }
751 mu.Lock()
752 writes = append(writes, host+"="+ver)
753 mu.Unlock()
754 return nil
755 })
756 w.publish("a", "v1") // the drain blocks inside this write
757 start := time.Now()
758 w.publish("a", "v2")
759 w.publish("b", "v1b")
760 w.publish("a", "v3")
761 if d := time.Since(start); d > time.Second {
762 t.Fatalf("publish waited %s on a slow write", d)
763 }
764 close(release1)
765 deadline := time.Now().Add(5 * time.Second)
766 for {
767 w.mu.Lock()
768 idle := !w.running
769 w.mu.Unlock()
770 if idle {
771 break
772 }
773 if time.Now().After(deadline) {
774 t.Fatal("the writer never drained")
775 }
776 time.Sleep(5 * time.Millisecond)
777 }
778 mu.Lock()
779 defer mu.Unlock()
780 var lastA string
781 for _, x := range writes {
782 if x[:2] == "a=" {
783 lastA = x
784 }
785 }
786 if lastA != "a=v3" || !slices.Contains(writes, "b=v1b") {
787 t.Fatalf("writes %v: want a's newest (v3) last and b's written", writes)
788 }
789 if slices.Contains(writes, "a=v2") {
790 t.Fatalf("writes %v: a stale intermediate version was written", writes)
791 }
792}
793
794// armAlways's callback reaches the registry, through the writer.
795func TestArmAlwaysRecordsTheServingVersion(t *testing.T) {
796 t.Parallel()
797 svc, reg, dd := rolloutFixture(t, false)
798 host := "bot.internal"
799 if err := reg.Upsert(registry.Route{Host: host, Kind: "sidecar", App: "blog", Account: "bab", Status: "ready"}); err != nil {
800 t.Fatal(err)
801 }
802 armAlways(svc, reg, dd)
803 svc.OnWorkerServing(host, release.StagedIdentity{Version: "v4"})
804 deadline := time.Now().Add(5 * time.Second)
805 for {
806 if rt, _ := reg.Get(host); rt != nil && rt.ServingVersion == "v4" {
807 return
808 }
809 if time.Now().After(deadline) {
810 t.Fatal("the serving version never reached the registry")
811 }
812 time.Sleep(5 * time.Millisecond)
813 }
814}
815
816// outageStore fails every read the way a bucket outage does.
817type outageStore struct{ release.Store }
818
819func (outageStore) Get(context.Context, string) ([]byte, string, error) {
820 return nil, "", errors.New("503 slow down")
821}
822
823// Carried obligation 1 (T3): the policy loader answers an error for any
824// store failure, and only a confirmed missing object as the defaults. A
825// policy never read is hold (rollout's policyCache).
826func TestThePolicyLoaderFailsOnAnyStoreError(t *testing.T) {
827 ctx := context.Background()
828 if m, err := rolloutPolicy(outageStore{})(ctx, "bab", "blog"); err == nil {
829 t.Fatalf("a store failure read as %+v", m)
830 }
831 s := release.NewDirStore(t.TempDir())
832 m, err := rolloutPolicy(s)(ctx, "bab", "blog")
833 if err != nil {
834 t.Fatalf("a missing config object: %v", err)
835 }
836 if got := rollout.PolicyFromManifest(m); got != rollout.DefaultPolicy() {
837 t.Fatalf("a missing config object read as %+v, want the defaults", got)
838 }
839 if _, err := s.Put(ctx, appconfig.Key("bab", "blog"), []byte("{not json"), ""); err != nil {
840 t.Fatal(err)
841 }
842 if m, err := rolloutPolicy(s)(ctx, "bab", "blog"); err == nil {
843 t.Fatalf("an undecodable config object read as %+v", m)
844 }
845}
846
847// A refusing box must not tell a member an awake managed instance was
848// restarted: the instance still runs its old config and nothing will restart
849// it (spec §8.1, an ack only once active). Its ack carries a fixed
850// RestartError and Restarted stays false. A sleeper reads the file on its
851// next wake, so its ack says delivered on wake (Hibernating, final review F7
852// (3)), never Restarted. An exec route the scheduler does not manage is
853// restarted as today and acked as restarted (F7 (2)).
854func TestRefusingConfigSyncDoesNotAckAnAwakeManagedHostAsRestarted(t *testing.T) {
855 ctx := context.Background()
856 svc, reg, dd := rolloutFixture(t, true)
857 _, pub, err := release.GenerateKey()
858 if err != nil {
859 t.Fatal(err)
860 }
861 noStore := func(context.Context) (release.Store, error) {
862 return nil, errors.New("no deployment store: secret detail")
863 }
864 if m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, noStore); m != nil {
865 t.Fatal("armed without a store")
866 }
867 store := release.NewDirStore(t.TempDir())
868 privText, _, err := appconfig.GenerateKey()
869 if err != nil {
870 t.Fatal(err)
871 }
872 priv, err := appconfig.ParsePrivateKey(privText, "test")
873 if err != nil {
874 t.Fatal(err)
875 }
876 if err := appconfig.SetEnv(ctx, store, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
877 t.Fatal(err)
878 }
879 for _, h := range []string{"awake.example.com", "asleep.example.com"} {
880 rt := execRoute(h, "bab", "eleven")
881 rt.Hibernate = true
882 if err := reg.Upsert(rt); err != nil {
883 t.Fatal(err)
884 }
885 }
886 // Always on, so not managed (rollout.Managed): today's restart.
887 if err := reg.Upsert(execRoute("plain.example.com", "bab", "eleven")); err != nil {
888 t.Fatal(err)
889 }
890 var restarted []string
891 cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{
892 awake: func(h string) bool { return h == "awake.example.com" },
893 restart: func(h string) error { restarted = append(restarted, h); return nil },
894 })
895 if !ok {
896 t.Fatal("refusing mode has no config callbacks")
897 }
898 acks := map[string]appconfig.ConfigAck{}
899 cb.Ack = func(a appconfig.ConfigAck) { acks[a.Host] = a }
900 if err := appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, t.TempDir(), "", cb); err != nil {
901 t.Fatal(err)
902 }
903 awake, ok := acks["awake.example.com"]
904 if !ok {
905 t.Fatalf("no ack for the awake host: %+v", acks)
906 }
907 if awake.Restarted || awake.RestartError != rolloutRestartRefused {
908 t.Fatalf("awake ack = %+v, want Restarted false and RestartError %q", awake, rolloutRestartRefused)
909 }
910 asleep, ok := acks["asleep.example.com"]
911 if !ok {
912 t.Fatalf("no ack for the sleeping host: %+v", acks)
913 }
914 if asleep.RestartError != "" || asleep.Restarted || !asleep.Hibernating {
915 t.Fatalf("sleeper ack = %+v, want delivered on wake: Hibernating, not restarted, no error", asleep)
916 }
917 if !slices.Equal(restarted, []string{"plain.example.com"}) {
918 t.Fatalf("today's restart ran for %v, want only the unmanaged route", restarted)
919 }
920 if plain := acks["plain.example.com"]; !plain.Restarted || plain.RestartError != "" || plain.Hibernating {
921 t.Fatalf("unmanaged ack = %+v, want today's: restarted", plain)
922 }
923}
924
925// Final review F7 (1) and (2), armed: config sync reports a managed native
926// sidecar's delivery to the scheduler, which owns its restart, and restarts
927// an exec route the scheduler does not manage as today (svc.Bounce, through
928// configLegs.restart), acking it as restarted at once; neither managed route
929// is restarted by config sync, and the unmanaged one is not reported.
930func TestArmedConfigSyncSplitsManagedFromUnmanaged(t *testing.T) {
931 ctx, cancel := context.WithCancel(context.Background())
932 t.Cleanup(cancel)
933 svc, reg, dd := rolloutFixture(t, true)
934 svc.GenerationsFile = release.GenerationsPath(dd)
935 _, pub, err := release.GenerateKey()
936 if err != nil {
937 t.Fatal(err)
938 }
939 managed := execRoute("m.example.com", "bab", "eleven")
940 managed.Hibernate, managed.Version = true, "v1"
941 bot := registry.Route{Host: "bot.eleven.bab.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bab", Status: "ready", Version: "v1"}
942 plain := execRoute("plain.example.com", "bab", "eleven") // always on: not managed
943 for _, rt := range []registry.Route{managed, bot, plain} {
944 if err := reg.Upsert(rt); err != nil {
945 t.Fatal(err)
946 }
947 }
948 if !rollout.Managed(managed) || !rollout.Managed(bot) || rollout.Managed(plain) {
949 t.Fatal("fixture: the routes are not managed as this test assumes")
950 }
951 store := release.NewDirStore(t.TempDir())
952 m := armRollout(ctx, overlapEnv(pub), svc, reg, dd, nil, func(context.Context) (release.Store, error) { return store, nil })
953 if m == nil {
954 t.Fatal("not armed")
955 }
956 privText, _, err := appconfig.GenerateKey()
957 if err != nil {
958 t.Fatal(err)
959 }
960 priv, err := appconfig.ParsePrivateKey(privText, "test")
961 if err != nil {
962 t.Fatal(err)
963 }
964 if err := appconfig.SetEnv(ctx, store, "bab", "eleven", map[string]string{"LOG_LEVEL": "debug"}); err != nil {
965 t.Fatal(err)
966 }
967 var restarted, delivered []string
968 cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{restart: func(h string) error { restarted = append(restarted, h); return nil }})
969 if !ok || cb.Delivered == nil {
970 t.Fatal("armed mode has no delivery callback")
971 }
972 report := cb.Delivered
973 cb.Delivered = func(h, rev string) { delivered = append(delivered, h); report(h, rev) }
974 acks := map[string]appconfig.ConfigAck{}
975 cb.Ack = m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks[a.Host] = a })
976 if err := appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, t.TempDir(), "", cb); err != nil {
977 t.Fatal(err)
978 }
979 // The manager's writer writes the acks (Task 11); Close writes what is
980 // still pending before it returns, so the map is complete after it.
981 m.Close(ctx)
982 slices.Sort(delivered)
983 if want := []string{bot.Host, managed.Host}; !slices.Equal(delivered, want) {
984 t.Fatalf("delivered %v, want %v: the managed sidecar's revision must reach the scheduler, the unmanaged route's must not", delivered, want)
985 }
986 if !slices.Equal(restarted, []string{plain.Host}) {
987 t.Fatalf("today's restart ran for %v, want only the unmanaged route", restarted)
988 }
989 if a, ok := acks[plain.Host]; !ok || !a.Restarted || a.RestartError != "" {
990 t.Fatalf("unmanaged ack = %+v (written %v), want today's: restarted, at once", a, ok)
991 }
992}
993
994// The awake test is the activator's: a managed host it owns. Without an
995// activator, or for a host it does not own, nothing is awake.
996func TestManagedAwakeNeedsTheActivatorToOwnAManagedHost(t *testing.T) {
997 svc, reg, _ := rolloutFixture(t, true)
998 rt := execRoute("m.example.com", "bab", "eleven")
999 rt.Hibernate = true
1000 if err := reg.Upsert(rt); err != nil {
1001 t.Fatal(err)
1002 }
1003 if managedAwake(nil, reg)("m.example.com") {
1004 t.Error("awake with no activator")
1005 }
1006 if managedAwake(svc, reg)("m.example.com") || managedAwake(svc, reg)("gone.example.com") {
1007 t.Error("awake though the activator owns nothing")
1008 }
1009}
1010