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 "context"
5 "crypto/ed25519"
6 "errors"
7 "os"
8 "path/filepath"
9 "strconv"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/provision"
15 "github.com/carlosframework/platform/internal/registry"
16 "github.com/carlosframework/platform/internal/release"
17 "github.com/carlosframework/platform/internal/rootstate"
18)
19
20type plan14RestartUnits struct {
21 props unitProps
22 exe string
23 opt string
24 effects []string
25 crashNew bool
26 badRollback bool
27 failNew bool
28 afterEffect func()
29 afterRestart func()
30}
31
32func (u *plan14RestartUnits) StartTime(pid int) (string, error) { return strconv.Itoa(pid), nil }
33
34func (u *plan14RestartUnits) Show(string) (unitProps, error) { return u.props, nil }
35func (u *plan14RestartUnits) Exe(int) (string, error) { return u.exe, nil }
36func (u *plan14RestartUnits) JournalTail(string, int) string { return "" }
37func (u *plan14RestartUnits) ResetFailed(unit string) error {
38 u.effects = append(u.effects, "reset "+unit)
39 return nil
40}
41func (u *plan14RestartUnits) TryRestart(unit string) error { return u.Restart(unit) }
42func (u *plan14RestartUnits) Restart(unit string) error {
43 u.effects = append(u.effects, "restart "+unit)
44 if u.afterRestart != nil {
45 u.afterRestart()
46 }
47 u.props.InvocationID += "x"
48 u.props.MainPID++
49 u.props.ActiveState, u.props.SubState = "active", "running"
50 var err error
51 u.exe, err = filepath.EvalSymlinks(release.LiveLink(u.opt, observedHost))
52 if u.crashNew && u.exe == release.VersionPath(u.opt, "bab", "v2", "console") {
53 u.props.ActiveState, u.props.SubState, u.props.MainPID = "failed", "failed", 0
54 }
55 if u.badRollback && u.exe == release.VersionPath(u.opt, "bab", "v1", "console") {
56 u.exe = release.VersionPath(u.opt, "bab", "v2", "console")
57 }
58 if u.failNew && u.exe == release.VersionPath(u.opt, "bab", "v2", "console") {
59 return os.ErrPermission
60 }
61 if u.afterEffect != nil {
62 u.afterEffect()
63 }
64 return err
65}
66
67func TestPlan14CrashedNamedUnitRollsBack(t *testing.T) {
68 t.Parallel()
69 e, _, units, c, pub := plan14AdoptionFixture(t)
70 units.crashNew = true
71 forward := rootRestartRouteWith(e.root.reg, units, units.opt, func(string) string { return "5s" }, e.root.obs.recordSettled(e.bake), c)
72 var rollbackErr error
73 rollback := rollbackUnitRestart(e.root.reg, units, units.opt, func(string) string { return "5s" }, c)
74 err := release.Pass(context.Background(), e.root.reg, e.store, pub, units.opt, forward,
75 release.BakeDir(e.bake), release.WithRestartFailurePropagation(),
76 release.WithRollbackRestart(func(host string) error { rollbackErr = rollback(host); return rollbackErr }))
77 if rollbackErr != nil || err == nil {
78 t.Fatalf("crash recovery: %v / %v", err, rollbackErr)
79 }
80 if len(units.effects) != 3 || units.props.MainPID <= 0 || units.exe != release.VersionPath(units.opt, "bab", "v1", "console") {
81 t.Fatalf("rollback was not verified: %+v", units)
82 }
83 if version, err := release.LiveVersion(units.opt, observedHost); err != nil || version != "v1" {
84 t.Fatalf("restored link = %q, %v", version, err)
85 }
86 if got := release.ReadHealthy(e.bake, "bab", "console"); len(got) != 1 {
87 t.Fatalf("crash recorded health: %+v", got)
88 }
89}
90
91func plan14AdoptionFixture(t *testing.T) (*proofRootEnv, registry.Route, *plan14RestartUnits, rootRestartChecks, ed25519.PublicKey) {
92 t.Helper()
93 e, rt, rec, source := plan14ProofFixture(t)
94 if got := e.root.observeAndRecord(rt, 15*time.Minute); got.err != nil {
95 t.Fatal(got.err)
96 }
97 source.healthy = func(acct, app, version, digest string) *release.HealthyBuild {
98 for _, b := range release.ReadHealthy(e.bake, acct, app) {
99 if b.SHA256 == digest {
100 return &b
101 }
102 }
103 return nil
104 }
105 u := &plan14RestartUnits{props: source.sys.(*fakeUnits).shows[0], exe: release.VersionPath(source.optDir, "bab", "v1", "console"), opt: source.optDir}
106 source.sys, e.root.obs.sys = u, u
107 e.root.obs.procExe = func(int) string { return u.exe }
108 rec.Pin = "v2"
109 plan14PutRecord(t, e.store, rec)
110 ctx := context.Background()
111 pub, priv, err := ed25519.GenerateKey(nil)
112 if err != nil {
113 t.Fatal(err)
114 }
115 for ver, body := range map[string]string{"v1": "x", "v2": "new build"} {
116 m := release.Manifest{App: "console", Version: ver, Kind: "binary", Artifacts: map[string]release.Artifact{release.AdoptPlatform(): {}}}
117 if err := release.Ship(ctx, e.store, "bab", m, map[string][]byte{release.AdoptPlatform(): []byte(body)}); err != nil {
118 t.Fatal(err)
119 }
120 }
121 if err := release.Promote(ctx, e.store, priv, "bab", "console", rt.Channel, "v2"); err != nil {
122 t.Fatal(err)
123 }
124 now, sleep := settleClock()
125 mem := &rollbackMemory{files: map[string][]byte{}}
126 c := rootRestartChecks{targetFile: os.Open, started: &sync.Map{}, attempts: func() (rootstate.Store, error) { return mem, nil }, rollback: source.rollback, prepareRollback: source.prepareRollback, resolve: source.resolve, host: source.forHost, ready: e.root.obs.ready, now: now, sleep: sleep}
127 return e, rt, u, c, pub
128}
129
130func TestPlan14AdoptionCanonicalReadiness(t *testing.T) {
131 t.Parallel()
132 mutations := map[string]func(*registry.Route){
133 "unchanged": func(*registry.Route) {},
134 "socket": func(r *registry.Route) { r.Socket = "/run/forged.sock" },
135 "addr": func(r *registry.Route) { r.Addr = "" },
136 "addrs": func(r *registry.Route) { r.Addrs = "127.0.0.1:1" },
137 "health": func(r *registry.Route) { r.Health = "" },
138 "unit": func(r *registry.Route) { r.Unit = "" },
139 "backing": func(r *registry.Route) { r.Backing = "exec" },
140 "empty upstream": func(r *registry.Route) { r.Socket, r.Addr, r.Addrs, r.Health = "", "", "", "" },
141 }
142 for name, mutate := range mutations {
143 t.Run(name, func(t *testing.T) {
144 e, rt, units, c, pub := plan14AdoptionFixture(t)
145 mutate(&rt)
146 if err := e.root.reg.Upsert(rt); err != nil {
147 t.Fatal(err)
148 }
149 c.ready = func(rt registry.Route) func() error {
150 if rt.Addr == "" && rt.Socket == "" {
151 return nil
152 }
153 return func() error { return os.ErrPermission }
154 }
155 e.root.obs.ready = c.ready
156 restart := rootRestartRouteWith(e.root.reg, units, units.opt, func(string) string { return "5s" }, e.root.obs.recordSettled(e.bake), c)
157 err := release.Pass(context.Background(), e.root.reg, e.store, pub, units.opt, restart,
158 release.BakeDir(e.bake), release.WithRestartFailurePropagation(),
159 release.WithRollbackRestart(rollbackUnitRestart(e.root.reg, units, units.opt, func(string) string { return "5s" }, c)))
160 if err == nil {
161 t.Fatal("adoption swallowed the canonical readiness refusal")
162 }
163 got, getErr := e.root.reg.Get(rt.Host)
164 if getErr != nil || got.Version == "v2" {
165 t.Fatalf("stamped adoption: %+v, %v", got, getErr)
166 }
167 if history := release.ReadHealthy(e.bake, "bab", "console"); len(history) != 1 || history[0].Version != "v1" {
168 t.Fatalf("new health: %+v", history)
169 }
170 })
171 }
172 t.Run("health write cannot block restart", func(t *testing.T) {
173 e, _, units, c, pub := plan14AdoptionFixture(t)
174 sentinel := errors.New("settled witness")
175 restart := rootRestartRouteWith(e.root.reg, units, units.opt, func(string) string { return "5s" }, func(registry.Route) error { return sentinel }, c)
176 err := release.Pass(context.Background(), e.root.reg, e.store, pub, units.opt, restart, release.WithRestartFailurePropagation())
177 if err != nil {
178 t.Fatalf("health write blocked restart: %v", err)
179 }
180 })
181 t.Run("positive adoption", func(t *testing.T) {
182 e, _, units, c, pub := plan14AdoptionFixture(t)
183 restart := rootRestartRouteWith(e.root.reg, units, units.opt, func(string) string { return "5s" }, e.root.obs.recordSettled(e.bake), c)
184 if err := release.Pass(context.Background(), e.root.reg, e.store, pub, units.opt, restart, release.WithRestartFailurePropagation()); err != nil {
185 t.Fatal(err)
186 }
187 rt, err := e.root.reg.Get(observedHost)
188 if err != nil || rt.Version != "v2" || len(release.ReadHealthy(e.bake, "bab", "console")) != 2 {
189 t.Fatalf("canonical adoption failed: %+v, %v", rt, err)
190 }
191 })
192}
193
194func TestPlan14RollbackCanonicalReadiness(t *testing.T) {
195 t.Parallel()
196 for _, attack := range []string{"missing", "unchanged", "account", "app", "unit", "backing", "socket", "addr", "addrs", "health", "all empty"} {
197 for _, ready := range []bool{false, true} {
198 name := attack + "/" + strconv.FormatBool(ready)
199 t.Run(name, func(t *testing.T) {
200 e, rt, units, c, pub := plan14AdoptionFixture(t)
201 c.ready = func(r registry.Route) func() error {
202 if r.Addr != "127.0.0.1:4000" || r.Health != rt.Health {
203 t.Fatalf("noncanonical rollback readiness: %+v", r)
204 }
205 return func() error {
206 if !ready || units.exe == release.VersionPath(units.opt, "bab", "v2", "console") {
207 return os.ErrPermission
208 }
209 return nil
210 }
211 }
212 forward := rootRestartRouteWith(e.root.reg, units, units.opt, func(string) string { return "5s" }, e.root.obs.recordSettled(e.bake), c)
213 restart := func(host string) error {
214 err := forward(host)
215 if attack == "missing" {
216 if e := e.root.reg.Delete(host); e != nil {
217 t.Fatal(e)
218 }
219 } else {
220 forged := rt
221 switch attack {
222 case "account":
223 forged.Account = "bap"
224 case "app":
225 forged.App = "other"
226 case "unit":
227 forged.Unit = "other.service"
228 case "backing":
229 forged.Backing = registry.BackingUnit
230 case "socket":
231 forged.Socket = "/run/forged.sock"
232 case "addr":
233 forged.Addr = ""
234 case "addrs":
235 forged.Addrs = "127.0.0.1:1"
236 case "health":
237 forged.Health = `{"path":"/forged"}`
238 case "all empty":
239 forged.Unit, forged.Backing, forged.Socket, forged.Addr, forged.Addrs, forged.Health = "", "", "", "", "", ""
240 }
241 if e := e.root.reg.Upsert(forged); e != nil {
242 t.Fatal(e)
243 }
244 }
245 return err
246 }
247 var rollbackErr error
248 rollback := rollbackUnitRestart(e.root.reg, units, units.opt, func(string) string { return "5s" }, c)
249 err := release.Pass(context.Background(), e.root.reg, e.store, pub, units.opt, restart,
250 release.BakeDir(e.bake), release.WithRestartFailurePropagation(),
251 release.WithRollbackRestart(func(host string) error { rollbackErr = rollback(host); return rollbackErr }))
252 if err == nil {
253 t.Fatal("forward readiness failure disappeared")
254 }
255 if ready != (rollbackErr == nil) {
256 t.Fatalf("rollback error = %v, readiness = %v", rollbackErr, ready)
257 }
258 if len(units.effects) != 3 {
259 t.Fatalf("rollback did not use the authoritative unit: %v", units.effects)
260 }
261 if !ready && !errors.Is(err, rollbackErr) {
262 t.Fatalf("rollback failure not reported: %v / %v", err, rollbackErr)
263 }
264 if history := release.ReadHealthy(e.bake, "bab", "console"); len(history) != 1 {
265 t.Fatalf("rollback wrote health: %+v", history)
266 }
267 })
268 }
269 }
270}
271
272func TestPlan14BindingChangesDuringRestart(t *testing.T) {
273 t.Parallel()
274 e, _, u, c, pub := plan14AdoptionFixture(t)
275 u.afterRestart = func() {
276 rec, _, err := provision.FindRecordByHost(context.Background(), e.store, observedHost)
277 if err != nil {
278 t.Fatal(err)
279 }
280 rec.RecordSeq++
281 plan14PutRecord(t, e.store, *rec)
282 }
283 forward := rootRestartRouteWith(e.root.reg, u, u.opt, func(string) string { return "5s" }, e.root.obs.recordSettled(e.bake), c)
284 err := release.Pass(context.Background(), e.root.reg, e.store, pub, u.opt, forward, release.BakeDir(e.bake), release.WithRestartFailurePropagation(), release.WithRollbackRestart(rollbackUnitRestart(e.root.reg, u, u.opt, func(string) string { return "5s" }, c)))
285 if err == nil || len(u.effects) != 1 || len(release.ReadHealthy(e.bake, "bab", "console")) != 1 {
286 t.Fatalf("changed binding settled: %v, %v", err, u.effects)
287 }
288}
289
290func TestPlan14RollbackInvocationChangesInProbe(t *testing.T) {
291 t.Parallel()
292 e, _, u, c, pub := plan14AdoptionFixture(t)
293 u.crashNew = true
294 c.ready = func(registry.Route) func() error { return func() error { u.props.InvocationID += "raced"; return nil } }
295 forward := rootRestartRouteWith(e.root.reg, u, u.opt, func(string) string { return "5s" }, nil, c)
296 var recovered error
297 err := release.Pass(context.Background(), e.root.reg, e.store, pub, u.opt, forward, release.BakeDir(e.bake), release.WithRestartFailurePropagation(), release.WithRollbackRestart(func(host string) error {
298 recovered = rollbackUnitRestart(e.root.reg, u, u.opt, func(string) string { return "5s" }, c)(host)
299 return recovered
300 }))
301 if err == nil || recovered == nil {
302 t.Fatal("changed invocation accepted as recovery")
303 }
304}
305