| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "io/fs" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "strings" |
| 14 | "sync" |
| 15 | "testing" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/hostactors" |
| 19 | "github.com/carlosframework/platform/internal/hostenv" |
| 20 | "github.com/carlosframework/platform/internal/hostfloor" |
| 21 | "github.com/carlosframework/platform/internal/lbhealth" |
| 22 | "github.com/carlosframework/platform/internal/status" |
| 23 | ) |
| 24 | |
| 25 | // stagedFacts walks a box through the joining stages, one per selectMode |
| 26 | // call: no invitation, invited, guard installed, guard-ready reported, |
| 27 | // then hosting. |
| 28 | type stagedFacts struct { |
| 29 | mu sync.Mutex |
| 30 | calls int |
| 31 | steps []joinFacts |
| 32 | } |
| 33 | |
| 34 | func (s *stagedFacts) facts() modeFacts { |
| 35 | pick := func() joinFacts { |
| 36 | s.mu.Lock() |
| 37 | defer s.mu.Unlock() |
| 38 | i := min(s.calls/5, len(s.steps)-1) // each stage for five checks |
| 39 | s.calls++ |
| 40 | return s.steps[i] |
| 41 | } |
| 42 | cur := joinFacts{} |
| 43 | f := modeFacts{box: "ada-hiking", now: time.Now} |
| 44 | f.enrollment = func() (hostfloor.Enrollment, error) { cur = pick(); return cur.facts().enrollment() } |
| 45 | f.roster = func(ctx context.Context) (hostactors.Roster, error) { return cur.facts().roster(ctx) } |
| 46 | f.guard = func() (hostfloor.Running, error) { return cur.facts().guard() } |
| 47 | f.invited = func(ctx context.Context, b string) (bool, error) { return cur.facts().invited(ctx, b) } |
| 48 | f.report = func(ctx context.Context, b string) (status.Box, error) { return cur.facts().report(ctx, b) } |
| 49 | return f |
| 50 | } |
| 51 | |
| 52 | func freePort(t *testing.T) string { |
| 53 | t.Helper() |
| 54 | l, err := net.Listen("tcp", "127.0.0.1:0") |
| 55 | if err != nil { |
| 56 | t.Skipf("no TCP listener in this sandbox: %v", err) |
| 57 | } |
| 58 | addr := l.Addr().String() |
| 59 | l.Close() |
| 60 | return addr |
| 61 | } |
| 62 | |
| 63 | // Joining publishes no edges file and admits nothing: it answers the load |
| 64 | // balancer 503 throughout, asks root to arm the guard once invited, |
| 65 | // publishes its report once armed, and returns as soon as it can host. |
| 66 | func TestJoiningServes503UntilItCanHostAndPublishesNoEdgesFile(t *testing.T) { |
| 67 | deploy, data := t.TempDir(), t.TempDir() |
| 68 | t.Setenv("CARLOS_DEPLOYMENT_DIR", deploy) |
| 69 | t.Setenv("CARLOS_DATA_DIR", data) |
| 70 | t.Cleanup(func() { lbhealth.Default.SetHosting(false) }) |
| 71 | addr := freePort(t) |
| 72 | f := enrolledFacts() |
| 73 | f.report = freshReport(time.Now()) |
| 74 | steps := &stagedFacts{steps: []joinFacts{{}, {invited: true}, guardInstalled(), func() joinFacts { g := guardInstalled(); g.report = freshReport(time.Now()); return g }(), f}} |
| 75 | j := productionJoining([]string{"-http", addr, "-https", ":0"}, steps.facts(), os.Getenv) |
| 76 | j.every = 20 * time.Millisecond |
| 77 | var mu sync.Mutex |
| 78 | var reports int |
| 79 | records := make(chan struct{}, 1) |
| 80 | j.record = func(context.Context) { records <- struct{}{} } |
| 81 | j.report = func(context.Context) error { mu.Lock(); reports++; mu.Unlock(); return nil } |
| 82 | j.logf = t.Logf |
| 83 | lbhealth.Default.SetHosting(true) // a stale flag from an earlier process state is cleared |
| 84 | done := make(chan modeDecision, 1) |
| 85 | go func() { d, _ := j.run(t.Context()); done <- d }() |
| 86 | // The health check answers 503, naming hosting, while joining. |
| 87 | var body string |
| 88 | for i := 0; i < 100; i++ { |
| 89 | resp, err := http.Get("http://" + addr + lbhealth.Path) |
| 90 | if err == nil { |
| 91 | b, _ := io.ReadAll(resp.Body) |
| 92 | resp.Body.Close() |
| 93 | if resp.StatusCode != http.StatusServiceUnavailable { |
| 94 | t.Fatalf("joining box answered %d", resp.StatusCode) |
| 95 | } |
| 96 | body = string(b) |
| 97 | break |
| 98 | } |
| 99 | time.Sleep(10 * time.Millisecond) |
| 100 | } |
| 101 | if !strings.Contains(body, lbhealth.CondHosting) { |
| 102 | t.Fatalf("health body %q", body) |
| 103 | } |
| 104 | if resp, err := http.Get("http://" + addr + "/"); err != nil || resp.StatusCode != http.StatusNotFound { |
| 105 | t.Fatalf("a joining box served something other than its health check: %v %v", resp, err) |
| 106 | } |
| 107 | select { |
| 108 | case d := <-done: |
| 109 | if d.Mode != modeHosting { |
| 110 | t.Fatalf("returned %+v", d) |
| 111 | } |
| 112 | case <-time.After(10 * time.Second): |
| 113 | t.Fatal("joining never returned") |
| 114 | } |
| 115 | <-records |
| 116 | if _, err := os.Stat(filepath.Join(data, joinRequestMarker)); err != nil { |
| 117 | t.Fatalf("root was never asked to arm the guard: %v", err) |
| 118 | } |
| 119 | mu.Lock() |
| 120 | if reports == 0 { |
| 121 | t.Fatal("the guard-ready report was never published") |
| 122 | } |
| 123 | mu.Unlock() |
| 124 | if lbhealth.Default.Hosting() { |
| 125 | t.Fatal("joining marked the box hosting") |
| 126 | } |
| 127 | err := filepath.WalkDir(deploy, func(p string, d fs.DirEntry, err error) error { |
| 128 | if err == nil && strings.Contains(filepath.ToSlash(p), "control/edges") { |
| 129 | t.Errorf("joining wrote %s", p) |
| 130 | } |
| 131 | return nil |
| 132 | }) |
| 133 | if err != nil { |
| 134 | t.Fatal(err) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func TestDecideAgentMode(t *testing.T) { |
| 139 | on := func(k string) string { |
| 140 | switch k { |
| 141 | case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA": |
| 142 | return "1" |
| 143 | } |
| 144 | return "" |
| 145 | } |
| 146 | noHA := func(k string) string { |
| 147 | if k == "CARLOS_HOST_CLAIM" { |
| 148 | return "1" |
| 149 | } |
| 150 | return "" |
| 151 | } |
| 152 | never := func(context.Context, func(string) string) (modeFacts, error) { |
| 153 | t.Fatal("facts read with joining mode off") |
| 154 | return modeFacts{}, nil |
| 155 | } |
| 156 | noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil } |
| 157 | quiet := func(string, ...any) {} |
| 158 | ctx := context.Background() |
| 159 | for name, env := range map[string]func(string) string{"claim off": func(string) string { return "" }, "region HA off": noHA} { |
| 160 | if !decideAgentMode(ctx, env, never, noJoin, time.Millisecond, quiet) { |
| 161 | t.Fatalf("%s: not hosting", name) |
| 162 | } |
| 163 | } |
| 164 | open := func(f joinFacts) func(context.Context, func(string) string) (modeFacts, error) { |
| 165 | return func(context.Context, func(string) string) (modeFacts, error) { return f.facts(), nil } |
| 166 | } |
| 167 | if !decideAgentMode(ctx, on, open(enrolledFacts()), noJoin, time.Millisecond, quiet) { |
| 168 | t.Fatal("an enrolled, armed box does not host") |
| 169 | } |
| 170 | var joined modeDecision |
| 171 | join := func(_ context.Context, _ modeFacts, d modeDecision) error { joined = d; return nil } |
| 172 | if decideAgentMode(ctx, on, open(joinFacts{invited: true}), join, time.Millisecond, quiet) || joined.Stage != stageInvited { |
| 173 | t.Fatalf("an invited box: joined %+v", joined) |
| 174 | } |
| 175 | // Unknown, then known: the gate waits, asks again, and logs the |
| 176 | // reason once. |
| 177 | reads, logs := 0, 0 |
| 178 | flaky := func(context.Context, func(string) string) (modeFacts, error) { |
| 179 | reads++ |
| 180 | f := enrolledFacts() |
| 181 | if reads < 3 { |
| 182 | f.rosterErr = errors.New("connection reset") |
| 183 | } |
| 184 | return f.facts(), nil |
| 185 | } |
| 186 | if !decideAgentMode(ctx, on, flaky, noJoin, time.Millisecond, func(string, ...any) { logs++ }) || reads != 3 || logs != 1 { |
| 187 | t.Fatalf("flaky roster: %d reads, %d log lines", reads, logs) |
| 188 | } |
| 189 | // Stopped while unknown: no claim agent. |
| 190 | stopped, cancel := context.WithCancel(ctx) |
| 191 | cancel() |
| 192 | failing := func(context.Context, func(string) string) (modeFacts, error) { |
| 193 | return modeFacts{}, errors.New("no bucket") |
| 194 | } |
| 195 | if decideAgentMode(stopped, on, failing, noJoin, time.Hour, quiet) { |
| 196 | t.Fatal("a stopped gate started the claim agent") |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | // A failed carlos system join is retried (Task 7's review): the path unit |
| 201 | // fires only when the marker changes, so joining touches it again once a |
| 202 | // minute for as long as the box stays invited with its guard not armed, |
| 203 | // and never more often. Once the guard is armed it stops asking. |
| 204 | func TestJoiningAsksRootAgainEachMinuteWhileInvited(t *testing.T) { |
| 205 | t.Cleanup(func() { lbhealth.Default.SetHosting(false) }) |
| 206 | var mu sync.Mutex |
| 207 | clock, checks := joinNow, 0 |
| 208 | var touches []time.Time |
| 209 | f := joinFacts{invited: true}.facts() |
| 210 | armedAfter := 40 // checks; then the guard is armed and not yet enrolled |
| 211 | f.enrollment = func() (hostfloor.Enrollment, error) { |
| 212 | mu.Lock() |
| 213 | defer mu.Unlock() |
| 214 | if checks >= armedAfter { |
| 215 | return guardInstalled().facts().enrollment() |
| 216 | } |
| 217 | return joinFacts{}.facts().enrollment() |
| 218 | } |
| 219 | f.roster = guardInstalled().facts().roster |
| 220 | f.guard = guardInstalled().facts().guard |
| 221 | f.now = func() time.Time { |
| 222 | mu.Lock() |
| 223 | defer mu.Unlock() |
| 224 | checks++ |
| 225 | clock = clock.Add(20 * time.Second) // each check, 20 seconds on |
| 226 | return clock |
| 227 | } |
| 228 | ctx, cancel := context.WithCancel(t.Context()) |
| 229 | defer cancel() |
| 230 | reported := make(chan struct{}, 1) |
| 231 | j := joining{ |
| 232 | facts: f, |
| 233 | every: time.Millisecond, |
| 234 | serve: func(ctx context.Context) error { <-ctx.Done(); return nil }, |
| 235 | record: func(context.Context) {}, |
| 236 | report: func(context.Context) error { |
| 237 | select { |
| 238 | case reported <- struct{}{}: |
| 239 | default: |
| 240 | } |
| 241 | return nil |
| 242 | }, |
| 243 | requestArm: func() error { |
| 244 | mu.Lock() |
| 245 | touches = append(touches, clock) |
| 246 | mu.Unlock() |
| 247 | return nil |
| 248 | }, |
| 249 | logf: t.Logf, |
| 250 | } |
| 251 | done := make(chan struct{}) |
| 252 | go func() { defer close(done); _, _ = j.run(ctx) }() |
| 253 | select { |
| 254 | case <-reported: |
| 255 | case <-time.After(10 * time.Second): |
| 256 | t.Fatal("the guard was armed and joining never published its report") |
| 257 | } |
| 258 | // A few more checks with the guard armed: no more touches. |
| 259 | for { |
| 260 | mu.Lock() |
| 261 | n := checks |
| 262 | mu.Unlock() |
| 263 | if n >= armedAfter+10 { |
| 264 | break |
| 265 | } |
| 266 | time.Sleep(time.Millisecond) |
| 267 | } |
| 268 | cancel() |
| 269 | <-done |
| 270 | mu.Lock() |
| 271 | defer mu.Unlock() |
| 272 | // 40 checks 20 seconds apart while invited: about 13 minutes, so a |
| 273 | // touch at the first check and then one each minute. |
| 274 | if len(touches) < 12 || len(touches) > 14 { |
| 275 | t.Fatalf("%d touches over %d invited checks: %v", len(touches), armedAfter, touches) |
| 276 | } |
| 277 | for i := 1; i < len(touches); i++ { |
| 278 | if gap := touches[i].Sub(touches[i-1]); gap < joinNudgeEvery || gap >= joinNudgeEvery+20*time.Second { |
| 279 | t.Fatalf("touch %d came %s after the one before; want once a minute", i, gap) |
| 280 | } |
| 281 | } |
| 282 | if last := touches[len(touches)-1]; last.After(joinNow.Add(time.Duration(armedAfter) * 20 * time.Second)) { |
| 283 | t.Fatalf("root was asked again at %s, after the guard was armed", last) |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | // Any error opening the facts (an empty group reference, say) is unknown: |
| 288 | // the gate waits and asks again, and neither joins nor hosts until a |
| 289 | // read answers (Task 6's review). |
| 290 | func TestDecideAgentModeWaitsOutAFactsError(t *testing.T) { |
| 291 | on := func(k string) string { |
| 292 | switch k { |
| 293 | case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA": |
| 294 | return "1" |
| 295 | } |
| 296 | return "" |
| 297 | } |
| 298 | for name, then := range map[string]joinFacts{"enrolled": enrolledFacts(), "invited": {invited: true}} { |
| 299 | t.Run(name, func(t *testing.T) { |
| 300 | reads, logs, joins := 0, 0, 0 |
| 301 | open := func(context.Context, func(string) string) (modeFacts, error) { |
| 302 | reads++ |
| 303 | if reads < 3 { |
| 304 | return modeFacts{}, errors.New("this box's group reference could not be read") |
| 305 | } |
| 306 | return then.facts(), nil |
| 307 | } |
| 308 | join := func(_ context.Context, _ modeFacts, d modeDecision) error { |
| 309 | if reads < 3 { |
| 310 | t.Fatalf("joined on read %d: %+v", reads, d) |
| 311 | } |
| 312 | joins++ |
| 313 | return nil |
| 314 | } |
| 315 | hosting := decideAgentMode(context.Background(), on, open, join, time.Millisecond, func(string, ...any) { logs++ }) |
| 316 | if reads != 3 || logs < 1 { |
| 317 | t.Fatalf("%d reads, %d log lines", reads, logs) |
| 318 | } |
| 319 | wantHosting, wantJoins := name == "enrolled", 1 |
| 320 | if wantHosting { |
| 321 | wantJoins = 0 |
| 322 | } |
| 323 | if hosting != wantHosting || joins != wantJoins { |
| 324 | t.Fatalf("hosting %t, %d joins; want %t, %d", hosting, joins, wantHosting, wantJoins) |
| 325 | } |
| 326 | }) |
| 327 | } |
| 328 | } |
| 329 | |
| 330 | // Task 6's review (R-A): a group box resolves its reference before the |
| 331 | // gate and before runAgent arms any loop, whatever the switches say, since |
| 332 | // those loops read metricsBoxLabel once and keep it. IMDS that fails twice |
| 333 | // holds everything back until the third read; every failure is logged with |
| 334 | // its cause. Any other box asks IMDS nothing and starts at once. |
| 335 | func TestAGroupBoxResolvesItsReferenceBeforeTheAgentStarts(t *testing.T) { |
| 336 | quick := func(int) time.Duration { return time.Millisecond } |
| 337 | never := func(context.Context, func(string) string) (modeFacts, error) { |
| 338 | t.Fatal("facts read with joining mode off") |
| 339 | return modeFacts{}, nil |
| 340 | } |
| 341 | enrolled := func(context.Context, func(string) string) (modeFacts, error) { return enrolledFacts().facts(), nil } |
| 342 | noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil } |
| 343 | for name, tc := range map[string]struct { |
| 344 | env map[string]string |
| 345 | facts func(context.Context, func(string) string) (modeFacts, error) |
| 346 | marked bool |
| 347 | }{ |
| 348 | "group box, switches off": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference}, facts: never, marked: true}, |
| 349 | "group box, switches on": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference, "CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1", "CARLOS_REGION": "eu-west-1"}, |
| 350 | facts: enrolled, marked: true}, |
| 351 | "other box, switches off": {env: map[string]string{}, facts: never}, |
| 352 | "other box, switches on": {env: map[string]string{"CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1"}, facts: enrolled}, |
| 353 | } { |
| 354 | t.Run(name, func(t *testing.T) { |
| 355 | withGroupIdentity(t, "", "", nil) |
| 356 | asked := 0 |
| 357 | groupIdentity = func() (string, string, error) { |
| 358 | asked++ |
| 359 | if asked < 3 { |
| 360 | return "", "", errors.New("IMDS answered 500") |
| 361 | } |
| 362 | return "123456789012", "i-0a61f580df1acfa65", nil |
| 363 | } |
| 364 | getenv := groupEnv(tc.env) |
| 365 | var lines []string |
| 366 | logf := func(format string, args ...any) { lines = append(lines, fmt.Sprintf(format, args...)) } |
| 367 | gatedAt, ranAt, label := -1, -1, "" |
| 368 | gate := func() bool { |
| 369 | gatedAt = asked |
| 370 | return decideAgentMode(context.Background(), getenv, tc.facts, noJoin, time.Millisecond, logf) |
| 371 | } |
| 372 | run := func() { ranAt = asked; label = metricsBoxLabel(getenv, "machine-1") } |
| 373 | agentEntry(getenv, resolveGroupBoxReference, quick, logf, gate, run) |
| 374 | want, ref := 0, status.Label("machine-1") |
| 375 | if tc.marked { |
| 376 | want, ref = 3, hostactors.GroupBoxReference("123456789012", "i-0a61f580df1acfa65") |
| 377 | } |
| 378 | if gatedAt != want || ranAt != want { |
| 379 | t.Fatalf("gate after %d IMDS reads, agent after %d; want %d", gatedAt, ranAt, want) |
| 380 | } |
| 381 | if label != ref { |
| 382 | t.Fatalf("the agent started with label %q, want %q", label, ref) |
| 383 | } |
| 384 | failures := 0 |
| 385 | for _, l := range lines { |
| 386 | if strings.Contains(l, "IMDS answered 500") { |
| 387 | failures++ |
| 388 | } |
| 389 | } |
| 390 | if tc.marked && failures != 2 { |
| 391 | t.Fatalf("logged %d failures with their cause, want 2: %q", failures, lines) |
| 392 | } |
| 393 | }) |
| 394 | } |
| 395 | } |
| 396 | |
| 397 | func TestAwaitGroupReferenceStopsWithItsContext(t *testing.T) { |
| 398 | ctx, cancel := context.WithCancel(context.Background()) |
| 399 | cancel() |
| 400 | failing := func() (string, error) { return "", errors.New("IMDS answered 500") } |
| 401 | if awaitGroupReference(ctx, failing, func(int) time.Duration { return time.Hour }, func(string, ...any) {}) { |
| 402 | t.Fatal("a stopped wait reported the reference resolved") |
| 403 | } |
| 404 | var got []time.Duration |
| 405 | for n := 1; n <= 8; n++ { |
| 406 | got = append(got, groupReferenceBackoff(n)) |
| 407 | } |
| 408 | want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 30 * time.Second, 30 * time.Second, 30 * time.Second} |
| 409 | if fmt.Sprint(got) != fmt.Sprint(want) { |
| 410 | t.Fatalf("backoff %v, want %v", got, want) |
| 411 | } |
| 412 | } |
| 413 | |
amadan