| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/hostbox" |
| 14 | "github.com/carlosframework/platform/internal/hostclaim" |
| 15 | "github.com/carlosframework/platform/internal/hostguard" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | ) |
| 19 | |
| 20 | // countingClaims counts the claim reads made through it, and runs onGet, |
| 21 | // when set, before each. |
| 22 | type countingClaims struct { |
| 23 | release.Store |
| 24 | mu sync.Mutex |
| 25 | claims int |
| 26 | onGet func(key string) |
| 27 | } |
| 28 | |
| 29 | func (s *countingClaims) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 30 | if strings.HasPrefix(key, "control/hosts/") { |
| 31 | s.mu.Lock() |
| 32 | s.claims++ |
| 33 | s.mu.Unlock() |
| 34 | } |
| 35 | if s.onGet != nil { |
| 36 | s.onGet(key) |
| 37 | } |
| 38 | return s.Store.Get(ctx, key) |
| 39 | } |
| 40 | |
| 41 | func (s *countingClaims) reads() int { |
| 42 | s.mu.Lock() |
| 43 | defer s.mu.Unlock() |
| 44 | n := s.claims |
| 45 | s.claims = 0 |
| 46 | return n |
| 47 | } |
| 48 | |
| 49 | func passRegistry(t *testing.T, hosts ...string) *registry.Registry { |
| 50 | t.Helper() |
| 51 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 52 | if err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | t.Cleanup(func() { reg.Close() }) |
| 56 | for _, host := range hosts { |
| 57 | if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil { |
| 58 | t.Fatal(err) |
| 59 | } |
| 60 | } |
| 61 | return reg |
| 62 | } |
| 63 | |
| 64 | // Review M2: the completion-only pass of a box with nothing to finish reads |
| 65 | // each of its claims once and asks nothing more: no identity read, however |
| 66 | // many rows. A pass that finds pending completions asks which box this is |
| 67 | // once, however many it finds. |
| 68 | func TestCompletionPassAsksNothingMoreWithNothingPending(t *testing.T) { |
| 69 | t.Parallel() |
| 70 | ctx := t.Context() |
| 71 | store := &countingClaims{Store: release.NewDirStore(t.TempDir())} |
| 72 | for _, host := range []string{"a.example", "b.example", "c.example"} { |
| 73 | passClaim(t, store.Store, host, hostclaim.Op{}) |
| 74 | } |
| 75 | completeOn(t, store.Store, "b.example", "rel-b", "dst", true) // finished already |
| 76 | reg := passRegistry(t, "a.example", "b.example", "c.example") |
| 77 | identities := 0 |
| 78 | local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| 79 | identities++ |
| 80 | return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| 81 | }}} |
| 82 | r, err := newCompletionRuntime(local, reg, &activator.Service{}) |
| 83 | if err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | store.reads() |
| 87 | r.once(ctx, store) |
| 88 | if n := store.reads(); identities != 0 || n != 3 { |
| 89 | t.Fatalf("nothing pending: %d identity reads, %d claim reads (want 0 and 3)", identities, n) |
| 90 | } |
| 91 | for _, host := range []string{"d.example", "e.example"} { |
| 92 | passClaim(t, store.Store, host, hostclaim.Op{}) |
| 93 | completeOn(t, store.Store, host, "rel-"+host[:1], "dst", false) |
| 94 | if err := reg.Upsert(registry.Route{Host: host, Status: "ready", Socket: "/run/" + host + ".sock"}); err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | } |
| 98 | f := &finishRecorder{} |
| 99 | r.Finish = f.finish |
| 100 | r.once(ctx, store) |
| 101 | if n := store.reads(); identities != 1 || len(f.calls) != 2 || n != 5 { |
| 102 | t.Fatalf("two pending: %d identity reads (want 1), finished %v, %d claim reads (want 5)", identities, f.calls, n) |
| 103 | } |
| 104 | } |
| 105 | |
| 106 | // Review M1: the completion-only pass reads the hosts its own op journals |
| 107 | // name beside its registry rows, so a host with a journal and no row is |
| 108 | // read; a journal it cannot read is said, and the rest are still read. |
| 109 | func TestCompletionPassReadsTheHostsItsJournalsName(t *testing.T) { |
| 110 | ctx := t.Context() |
| 111 | store := release.NewDirStore(t.TempDir()) |
| 112 | passClaim(t, store, "row.example", hostclaim.Op{}) |
| 113 | passClaim(t, store, "rowless.example", hostclaim.Op{}) |
| 114 | completeOn(t, store, "rowless.example", "rel-1", "dst", false) |
| 115 | data := t.TempDir() |
| 116 | marker, err := relocationProbeFencePath(data, "rel-1") |
| 117 | if err != nil { |
| 118 | t.Fatal(err) |
| 119 | } |
| 120 | if _, err := relocationProbeFenceOwner(marker, "rowless.example", 1); err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | bad := filepath.Join(data, "relocate", "ops", "rel-2", "attempts.json") |
| 124 | if err := os.MkdirAll(filepath.Dir(bad), 0o700); err != nil { |
| 125 | t.Fatal(err) |
| 126 | } |
| 127 | if err := os.WriteFile(bad, []byte("not a journal"), 0o600); err != nil { |
| 128 | t.Fatal(err) |
| 129 | } |
| 130 | reg := passRegistry(t, "row.example") |
| 131 | local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: data}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| 132 | return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| 133 | }}} |
| 134 | r, err := newCompletionRuntime(local, reg, &activator.Service{}) |
| 135 | if err != nil { |
| 136 | t.Fatal(err) |
| 137 | } |
| 138 | hosts, err := r.Hosts() |
| 139 | if len(hosts) != 2 || hosts[0] != "row.example" || hosts[1] != "rowless.example" || err == nil { |
| 140 | t.Fatalf("hosts %v, err %v: want the row, the journal's host and the unreadable journal", hosts, err) |
| 141 | } |
| 142 | buf := captureRacyLog(t) |
| 143 | f := &finishRecorder{} |
| 144 | r.Finish = f.finish |
| 145 | r.once(ctx, store) |
| 146 | if len(f.calls) != 1 || f.calls[0] != "rowless.example:rel-1" || !strings.Contains(buf.String(), "relocate: list hosts:") { |
| 147 | t.Fatalf("finished %v, log:\n%s", f.calls, buf.String()) |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | // Review M8: an alert is one per host, step and cause. Two steps failing |
| 152 | // with one cause are both said; a cause that clears and returns is said |
| 153 | // again; a host the pass no longer reads is forgotten. |
| 154 | func TestRelocatePassAlertsPerStepAndForgetsClearedCauses(t *testing.T) { |
| 155 | ctx := t.Context() |
| 156 | store := release.NewDirStore(t.TempDir()) |
| 157 | passClaim(t, store, "a.example", relocatingOp("rel-a")) |
| 158 | same := errors.New("the roster could not be read") |
| 159 | var order []string |
| 160 | participant := &flakyStepper{name: "participant", order: &order, err: same} |
| 161 | r := &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order, err: same}, Participant: participant, Box: "dst"} |
| 162 | listed := []string{"a.example"} |
| 163 | r.Hosts = func() ([]string, error) { return listed, nil } |
| 164 | buf := captureRacyLog(t) |
| 165 | r.once(ctx, store) |
| 166 | r.once(ctx, store) |
| 167 | if strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()) != 1 || strings.Count(buf.String(), "relocate a.example: worker: "+same.Error()) != 1 { |
| 168 | t.Fatalf("not one alert per step:\n%s", buf.String()) |
| 169 | } |
| 170 | participant.err = nil // the cause clears |
| 171 | r.once(ctx, store) |
| 172 | participant.err = same // and returns |
| 173 | r.once(ctx, store) |
| 174 | if n := strings.Count(buf.String(), "relocate a.example: participant: "+same.Error()); n != 2 { |
| 175 | t.Fatalf("a returning cause said %d times, want twice:\n%s", n, buf.String()) |
| 176 | } |
| 177 | if r.alerts.size() != 2 { |
| 178 | t.Fatalf("held %d causes, want 2", r.alerts.size()) |
| 179 | } |
| 180 | listed = nil |
| 181 | r.once(ctx, store) |
| 182 | if r.alerts.size() != 0 { |
| 183 | t.Fatalf("held %d causes for hosts no longer read", r.alerts.size()) |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | // flakyStepper is a stepRecorder whose error can change between passes. |
| 188 | type flakyStepper struct { |
| 189 | name string |
| 190 | order *[]string |
| 191 | err error |
| 192 | } |
| 193 | |
| 194 | func (s *flakyStepper) Step(_ context.Context, host string) (hostclaim.Observed, error) { |
| 195 | *s.order = append(*s.order, s.name+":"+host) |
| 196 | return hostclaim.Observed{}, s.err |
| 197 | } |
| 198 | |
| 199 | // Review M6: the sweep reads every claim before it takes the journals' |
| 200 | // lock exclusively, so a wake's journal write never waits on a bucket |
| 201 | // read; and what it decided unlocked it removes only as it finds it under |
| 202 | // the lock. A directory that held no journal when it was read, and holds |
| 203 | // one by the time the lock is held, is kept. |
| 204 | func TestSweepReadsClaimsUnlockedAndChecksAgainUnderTheLock(t *testing.T) { |
| 205 | ctx := t.Context() |
| 206 | data := t.TempDir() |
| 207 | ops := filepath.Join(data, "relocate", "ops") |
| 208 | for _, dir := range []string{"aaa", "bbb"} { |
| 209 | if err := os.MkdirAll(filepath.Join(ops, dir), 0o700); err != nil { |
| 210 | t.Fatal(err) |
| 211 | } |
| 212 | } |
| 213 | // aaa: a write's temporary only. bbb: b.example's journal, its claim gone. |
| 214 | if err := os.WriteFile(filepath.Join(ops, "aaa", ".attempts.json.1"), []byte("{"), 0o600); err != nil { |
| 215 | t.Fatal(err) |
| 216 | } |
| 217 | if err := os.WriteFile(filepath.Join(ops, "bbb", "install.json"), []byte(`{"v":1,"host":"b.example"}`), 0o600); err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | locked := false |
| 221 | store := &countingClaims{Store: release.NewDirStore(t.TempDir())} |
| 222 | store.onGet = func(key string) { |
| 223 | if relocationJournals.TryRLock() { |
| 224 | relocationJournals.RUnlock() |
| 225 | } else { |
| 226 | locked = true |
| 227 | } |
| 228 | // A write lands in aaa after the sweep read it as empty. |
| 229 | if key == "control/hosts/b.example.json" { |
| 230 | if err := os.WriteFile(filepath.Join(ops, "aaa", "attempts.json"), []byte(`{"v":1,"host":"a.example","op":"aaa"}`), 0o600); err != nil { |
| 231 | t.Error(err) |
| 232 | } |
| 233 | } |
| 234 | } |
| 235 | if err := sweepRelocationJournals(ctx, data, store); err != nil { |
| 236 | t.Fatal(err) |
| 237 | } |
| 238 | if locked { |
| 239 | t.Fatal("a claim was read under the journals' exclusive lock") |
| 240 | } |
| 241 | if store.reads() == 0 { |
| 242 | t.Fatal("the sweep read no claim") |
| 243 | } |
| 244 | if _, err := os.Stat(filepath.Join(ops, "bbb")); !errors.Is(err, os.ErrNotExist) { |
| 245 | t.Fatalf("the ended op's journals survived: %v", err) |
| 246 | } |
| 247 | if _, err := os.Stat(filepath.Join(ops, "aaa", "attempts.json")); err != nil { |
| 248 | t.Fatalf("a journal committed after the decision was swept: %v", err) |
| 249 | } |
| 250 | } |
| 251 | |
| 252 | // closingStepper closes stepped on its first Step. |
| 253 | type closingStepper struct { |
| 254 | once *sync.Once |
| 255 | stepped chan struct{} |
| 256 | } |
| 257 | |
| 258 | func (s closingStepper) Step(context.Context, string) (hostclaim.Observed, error) { |
| 259 | s.once.Do(func() { close(s.stepped) }) |
| 260 | return hostclaim.Observed{}, nil |
| 261 | } |
| 262 | |
| 263 | // Review M3 (run with -race): the pass's loop owns the runtime's fields |
| 264 | // once it starts, and its first pass replaces them with the full pass's |
| 265 | // (tryUpgrade). The launch says what it started from what it decided |
| 266 | // before the loop, never by reading the runtime after it. |
| 267 | func TestLaunchRelocationPassReadsNothingTheLoopOwns(t *testing.T) { |
| 268 | store := release.NewDirStore(t.TempDir()) |
| 269 | passClaim(t, store, "a.example", relocatingOp("rel-a")) |
| 270 | reg := passRegistry(t) |
| 271 | local := &productionClaimLocal{legacyParticipation: true, assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| 272 | return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| 273 | }}} |
| 274 | stepped := make(chan struct{}) |
| 275 | step := closingStepper{once: &sync.Once{}, stepped: stepped} |
| 276 | r, err := relocationPassFor(local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) { |
| 277 | return &RelocationRuntime{Worker: step, Participant: step, Box: "dst", Finish: func(context.Context, string, string) error { return nil }}, nil |
| 278 | }) |
| 279 | if err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | buf := captureRacyLog(t) |
| 283 | ctx, cancel := context.WithCancel(t.Context()) |
| 284 | done := launchRelocationPass(ctx, r, store, true) |
| 285 | <-stepped |
| 286 | cancel() |
| 287 | <-done |
| 288 | if !strings.Contains(buf.String(), "relocation pass started (completions only: false)") || !strings.Contains(buf.String(), "agent: region HA: relocation pass started") { |
| 289 | t.Fatalf("not said:\n%s", buf.String()) |
| 290 | } |
| 291 | } |
| 292 | |
amadan