amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package host
4
5import (
6 "bytes"
7 "context"
8 "os"
9 "path/filepath"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/activator"
15 "github.com/carlosframework/platform/internal/hostclaim"
16 "github.com/carlosframework/platform/internal/hostfiles"
17 "github.com/carlosframework/platform/internal/lease"
18 "github.com/carlosframework/platform/internal/lease/leasetest"
19 "github.com/carlosframework/platform/internal/provision"
20 "github.com/carlosframework/platform/internal/registry"
21 "github.com/carlosframework/platform/internal/release"
22)
23
24func runFilesSeedRealPreparation(t *testing.T, p *servingPreparation, live ...filesSeedLiveWitness) {
25 t.Helper()
26 f := p.f
27 f.actor.Bucket = f.f.bucket
28 f.actor.Cloud.Region = "eu-west-1"
29 f.actor.HomeRegion = "ie"
30 f.actor.FleetHomeRegion = "ie"
31 p.local.Actor = f.actor
32 rt := registry.Route{Host: f.f.hostName(), Kind: "instance", Hibernate: true, Account: "bab", App: "jam", DBPath: f.db, Socket: provision.NestedSocket(f.f.hostName())}
33 reg, err := registry.Open(filepath.Join(t.TempDir(), "seed.db"))
34 if err != nil {
35 t.Fatal(err)
36 }
37 t.Cleanup(func() { reg.Close() })
38 if err := reg.Upsert(rt); err != nil {
39 t.Fatal(err)
40 }
41 if err := reg.SetNextDue(rt.Host, 777); err != nil {
42 t.Fatal(err)
43 }
44 filesDir := filepath.Join(filepath.Dir(f.db), rt.Host)
45 if err := os.MkdirAll(filesDir, 0700); err != nil {
46 t.Fatal(err)
47 }
48 if err := os.WriteFile(filepath.Join(filesDir, "payload"), []byte("real seeded bytes"), 0600); err != nil {
49 t.Fatal(err)
50 }
51 if err := os.WriteFile(filepath.Join(filepath.Dir(f.db), rt.Host+".env"), []byte("KEY=value\n"), 0600); err != nil {
52 t.Fatal(err)
53 }
54 source := &movesFilesSource{DataDir: f.f.guard.DataDir, Routes: reg, newFiles: func(context.Context, string, string) (release.Store, error) {
55 return casBucketStore{b: f.f.data.(*leasetest.VersionedBucket), bucket: f.f.bucket}, nil
56 }}
57 a := p.assembly
58 a.Data = f.f.data.(*leasetest.VersionedBucket)
59 a.Identity = handoverRecoveryIdentity(t, f.actor)
60 a.Wake.seedFactory = func(ctx context.Context, op *filesSeedOperation, token *activator.SeedExclusion) (*activator.ClaimRuntime, *ClaimWakeAssembly, error) {
61 runtime, wake, err := newFilesSeedRuntime(ctx, a, p.source, p.writer, op, token)
62 if err != nil {
63 return runtime, wake, err
64 }
65 if f.f.guard.DataDir != "/data" {
66 // Local fixture omits only installed /data launch-shape preflight.
67 runtime.Prerequisites = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, policy activator.StartPolicy) error {
68 if err := wake.ServingPrerequisites(ctx, rt, o); err != nil {
69 return err
70 }
71 in, err := wake.Inputs.Select(ctx, rt, o, policy)
72 if err != nil {
73 return err
74 }
75 if err := wake.stopPredecessor(ctx, o); err != nil {
76 return err
77 }
78 held, err := lease.Acquire(ctx, f.f.bucket, o.Claim.Holder.Location, wake.LeaseRegion)
79 if err != nil {
80 return err
81 }
82 live, cancel := context.WithCancel(ctx)
83 done := make(chan struct{})
84 go func() { <-live.Done(); close(done) }()
85 wake.pending[rt.Host] = &claimWakeAttempt{owner: wake, observed: o, inputs: in, held: held, ctx: live, cancel: cancel, done: done}
86 return nil
87 }
88 }
89 return runtime, wake, nil
90 }
91 openedSelection, err := source.Open(t.Context(), f.f.claims, f.f.data.(*leasetest.VersionedBucket), f.actor)
92 if err != nil {
93 t.Fatal(err)
94 }
95 classification, err := inspectFilesSeed(t.Context(), f.f.claims, openedSelection.(*movesFiles), f.f.read(), f.actor, &rt)
96 if err != nil {
97 t.Fatal("classification", err)
98 }
99 if _, err := selectFilesSeedInputs(t.Context(), p.owners, rt, classification); err != nil {
100 t.Fatal("input selection", err)
101 }
102 executor := &filesSeedExecutor{assembly: &a, service: &activator.Service{}, registry: reg, source: source}
103 for _, witness := range live {
104 if witness.seed != nil {
105 witness.seed(t, p, executor)
106 return
107 }
108 }
109 var output bytes.Buffer
110 if err := executor.run(t.Context(), rt.Host, &output); err != nil {
111 t.Fatalf("seed: %v\n%s", err, output.String())
112 }
113 seeded := f.f.read()
114 if seeded.Claim.Origins == nil || seeded.Claim.Holder.WakeRetry.AttemptsReserved != 1 || seeded.Claim.Holder.WakeRetry.State != "open" {
115 t.Fatal("seed did not settle one real admission", seeded)
116 }
117 if len(executor.pending) != 0 {
118 t.Fatal("seed retained its exclusion")
119 }
120 before := seeded.ETag
121 if err := executor.run(t.Context(), rt.Host, &output); err != nil {
122 t.Fatal("established rerun", err)
123 }
124 if f.f.read().ETag != before {
125 t.Fatal("established rerun mutated claim")
126 }
127 opened, err := source.Open(t.Context(), f.f.claims, f.f.data.(*leasetest.VersionedBucket), f.actor)
128 if err != nil {
129 t.Fatal(err)
130 }
131 ordinary := opened.(*movesFiles)
132 for wake := 0; wake < 2; wake++ {
133 f.f.now = f.f.now.Add(time.Minute)
134 normal := newServingPreparation(t, f, "boot", "controller", func(w *ClaimWakeAssembly) {
135 w.Inputs.(*claimServingOwners).Sources.Files = ordinary
136 w.ServingPrerequisites = servingFilesPrerequisites(f.f.claims, ordinary)
137 for _, witness := range live {
138 witness.configure(w)
139 }
140 })
141 for _, witness := range live {
142 if witness.prepare != nil {
143 witness.prepare(normal)
144 }
145 }
146 admitted, start := normal.prepare(t)
147 if err := start.Restore(t.Context()); err != nil {
148 t.Fatal("cold restore", wake, err)
149 }
150 if err := start.Reserve(t.Context(), 0); err != nil {
151 t.Fatal(err)
152 }
153 if err := start.Materialize(t.Context()); err != nil {
154 t.Fatal("ordinary C8", wake, err)
155 }
156 prefix := movesFilesPrefix(admitted.Claim.Holder.Location, admitted.Claim.Holder.Generation)
157 manifest, found, err := hostfiles.Load(t.Context(), ordinary.Files, prefix)
158 if err != nil || !found {
159 t.Fatal(err)
160 }
161 if _, ok := manifest.Entries["payload"]; !ok {
162 t.Fatal("files lost at cold wake")
163 }
164 b, err := os.ReadFile(filepath.Join(filesDir, "payload"))
165 if err != nil || string(b) != "real seeded bytes" {
166 t.Fatal("restored files", string(b), err)
167 }
168 for _, witness := range live {
169 witness.serve(t, normal, admitted, start)
170 }
171 healthy := false
172 for _, witness := range live {
173 healthy = healthy || witness.healthy
174 }
175 if !healthy {
176 if err := start.Failed(t.Context(), 0); err != nil {
177 t.Fatal(err)
178 }
179 }
180 if err := normal.runtime.FailedAdmission(t.Context(), admitted, hostclaim.ErrWrongState); err != nil {
181 t.Fatal(err)
182 }
183 if ordinary.Seed != nil {
184 t.Fatal("ordinary provider armed")
185 }
186 }
187 healthy := false
188 for _, witness := range live {
189 healthy = healthy || witness.healthy
190 }
191 if healthy {
192 if f.f.read().Claim.Holder.WakeRetry != nil {
193 t.Fatal("health did not clear retry debt")
194 }
195 } else if f.f.read().Claim.Holder.WakeRetry.AttemptsReserved != 3 {
196 t.Fatal("attempt budget changed")
197 }
198 t.Log(output.String())
199}
200
201type filesSeedLiveWitness struct {
202 host string
203 shared bool
204 start func(*testing.T, *servingPreparation, hostclaim.Observed, activator.ClaimStart) func()
205 healthy bool
206 prepare func(*servingPreparation)
207 seed func(*testing.T, *servingPreparation, *filesSeedExecutor)
208 configure func(*ClaimWakeAssembly)
209 serve func(*testing.T, *servingPreparation, hostclaim.Observed, activator.ClaimStart)
210}
211
212func TestFilesSeedRealOriginPreparation(t *testing.T) {
213 runFilesSeedOrdinaryPreparation(t, "real-seed", "same-controller")
214}
215
216func TestFilesSeedReportsSkippedPaths(t *testing.T) {
217 witness := filesSeedLiveWitness{seed: func(t *testing.T, p *servingPreparation, executor *filesSeedExecutor) {
218 host := p.f.f.hostName()
219 dir := filepath.Join(filepath.Dir(p.f.db), host)
220 large, err := os.Create(filepath.Join(dir, "large.bin"))
221 if err != nil {
222 t.Fatal(err)
223 }
224 err = large.Truncate(hostfiles.MaxFileBytes + 1)
225 large.Close()
226 if err != nil {
227 t.Fatal(err)
228 }
229 if err := os.Symlink("payload", filepath.Join(dir, "link\nname")); err != nil {
230 t.Fatal(err)
231 }
232 var output bytes.Buffer
233 if err := executor.run(t.Context(), host, &output); err != nil {
234 t.Fatalf("seed: %v\n%s", err, output.String())
235 }
236 for _, want := range []string{`skipped file "large.bin": exceeds 64 MiB`, `skipped file "link\nname": not a regular file.`, `: seeded.`} {
237 if !strings.Contains(output.String(), want) {
238 t.Fatalf("missing %q in output:\n%s", want, output.String())
239 }
240 }
241 }}
242 runFilesSeedPreparationWithLive(t, "real-seed", "same-controller", nil, []filesSeedLiveWitness{witness})
243}
244
245func TestFilesSeedPreparationChangedController(t *testing.T) {
246 witness := filesSeedLiveWitness{configure: func(*ClaimWakeAssembly) {}, serve: func(*testing.T, *servingPreparation, hostclaim.Observed, activator.ClaimStart) {}}
247 witness.prepare = func(p *servingPreparation) {
248 p.local.ControllerID, p.local.PID, p.local.ProcessStart = "new-controller", 88, "new-start"
249 p.f.f.peer = StopPeer{UID: 2001, PID: 88, ProcessStart: "new-start", BootID: p.local.BootID, ControllerID: "new-controller"}
250 p.f.p.root.Stop.Coordinator.CarlosUID = 2001
251 p.wake.Retries.BootID, p.wake.Retries.ControllerID = p.local.BootID, "new-controller"
252 if len(p.wake.pending) != 0 {
253 t.Fatal("rebuilding active fixture")
254 }
255 p.wake.pending = nil
256 var err error
257 p.runtime, err = p.wake.Build(t.Context(), p.source, p.writer)
258 if err != nil {
259 t.Fatal(err)
260 }
261 }
262 runFilesSeedPreparationWithLive(t, "real-seed", "same-controller", nil, []filesSeedLiveWitness{witness})
263}
264