| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "net/http" |
| 6 | "net/http/httptest" |
| 7 | "sync" |
| 8 | "sync/atomic" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/provision" |
| 13 | ) |
| 14 | |
| 15 | // waitFor polls cond until it holds or within elapses. Every prober assertion |
| 16 | // is time-based (a probe tick has to land before state can move), so the |
| 17 | // windows here are deliberately generous — an order of magnitude above the |
| 18 | // tick they wait on — rather than exact: a loaded CI box must not turn a |
| 19 | // correct prober into a red build. |
| 20 | func waitFor(t *testing.T, within time.Duration, what string, cond func() bool) { |
| 21 | t.Helper() |
| 22 | deadline := time.Now().Add(within) |
| 23 | for time.Now().Before(deadline) { |
| 24 | if cond() { |
| 25 | return |
| 26 | } |
| 27 | time.Sleep(5 * time.Millisecond) |
| 28 | } |
| 29 | if !cond() { |
| 30 | t.Fatalf("timed out after %s waiting for %s", within, what) |
| 31 | } |
| 32 | } |
| 33 | |
| 34 | // switchableBackend is an httptest server whose answer flips between the |
| 35 | // configured status and 200 under an atomic flag, plus a hit counter. |
| 36 | // |
| 37 | // The flag exists because the obvious alternative — reassigning |
| 38 | // srv.Config.Handler once the server is live — is a data race: net/http reads |
| 39 | // Server.Handler on every connection it serves, so mutating it while the |
| 40 | // prober is mid-tick is exactly the kind of race `-race` exists to catch. |
| 41 | func switchableBackend(t *testing.T, downStatus int) (*httptest.Server, *atomic.Bool, *atomic.Int64) { |
| 42 | t.Helper() |
| 43 | ok := new(atomic.Bool) |
| 44 | hits := new(atomic.Int64) |
| 45 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 46 | hits.Add(1) |
| 47 | if ok.Load() { |
| 48 | w.WriteHeader(http.StatusOK) |
| 49 | return |
| 50 | } |
| 51 | w.WriteHeader(downStatus) |
| 52 | })) |
| 53 | t.Cleanup(ts.Close) |
| 54 | return ts, ok, hits |
| 55 | } |
| 56 | |
| 57 | func TestProberFailsAwayAndBack(t *testing.T) { |
| 58 | healthy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 59 | w.WriteHeader(http.StatusOK) |
| 60 | })) |
| 61 | t.Cleanup(healthy.Close) |
| 62 | flaky, flakyOK, _ := switchableBackend(t, http.StatusInternalServerError) |
| 63 | |
| 64 | healthyAddr := healthy.Listener.Addr().String() |
| 65 | flakyAddr := flaky.Listener.Addr().String() |
| 66 | spec := probeSpec{ |
| 67 | Host: "m.example.com", |
| 68 | Addrs: []string{healthyAddr, flakyAddr}, |
| 69 | Health: provision.Health{ |
| 70 | Path: "/", Status: 200, IntervalMS: 20, FailAfter: 3, RiseAfter: 2, |
| 71 | }.WithDefaults(), |
| 72 | } |
| 73 | |
| 74 | p := newProberSet() |
| 75 | defer p.Stop() |
| 76 | // Optimism: a fresh Sync must not 503 a healthy fleet while the first |
| 77 | // probes run, so both upstreams are live before any probe has landed. |
| 78 | p.Sync(map[string]probeSpec{"m.example.com": spec}) |
| 79 | if live := p.Live("m.example.com"); len(live) != 2 { |
| 80 | t.Fatalf("live immediately after Sync = %v, want both upstreams (all start live)", live) |
| 81 | } |
| 82 | |
| 83 | waitFor(t, 2*time.Second, "the 500ing upstream to fail away", func() bool { |
| 84 | return len(p.Live("m.example.com")) == 1 |
| 85 | }) |
| 86 | if live := p.Live("m.example.com"); live[0] != healthyAddr { |
| 87 | t.Fatalf("live = %v, want only the healthy upstream %s", live, healthyAddr) |
| 88 | } |
| 89 | st := p.State()["m.example.com"] |
| 90 | if len(st) != 2 { |
| 91 | t.Fatalf("State() = %v, want one entry per upstream", st) |
| 92 | } |
| 93 | down := st[1] |
| 94 | if down.Addr != flakyAddr || down.Up { |
| 95 | t.Fatalf("State()[1] = %+v, want %s down", down, flakyAddr) |
| 96 | } |
| 97 | if down.Fails < 3 { |
| 98 | t.Errorf("down upstream Fails = %d, want >= FailAfter (3)", down.Fails) |
| 99 | } |
| 100 | if down.LastError == "" { |
| 101 | t.Error("down upstream LastError is empty, want the wrong-status reason") |
| 102 | } |
| 103 | if down.Since.IsZero() { |
| 104 | t.Error("down upstream Since is zero, want the transition time") |
| 105 | } |
| 106 | |
| 107 | // The flaky upstream recovers → rejoins after RiseAfter passes. |
| 108 | flakyOK.Store(true) |
| 109 | waitFor(t, 2*time.Second, "the recovered upstream to rejoin", func() bool { |
| 110 | return len(p.Live("m.example.com")) == 2 |
| 111 | }) |
| 112 | if got := p.State()["m.example.com"][1]; !got.Up || got.Fails != 0 || got.LastError != "" { |
| 113 | t.Errorf("recovered upstream = %+v, want up with cleared fails/error", got) |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | func TestProberSendsConfiguredHostHeader(t *testing.T) { |
| 118 | var mu sync.Mutex |
| 119 | var seenHost, seenPath string |
| 120 | ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 121 | mu.Lock() |
| 122 | seenHost, seenPath = r.Host, r.URL.Path |
| 123 | mu.Unlock() |
| 124 | w.WriteHeader(http.StatusOK) |
| 125 | })) |
| 126 | t.Cleanup(ts.Close) |
| 127 | saw := func() (string, string) { |
| 128 | mu.Lock() |
| 129 | defer mu.Unlock() |
| 130 | return seenHost, seenPath |
| 131 | } |
| 132 | |
| 133 | // HostHeader set: it wins over the route host, so a shared upstream that |
| 134 | // vhosts on Host can be probed on the name it actually answers to. |
| 135 | p := newProberSet() |
| 136 | p.Sync(map[string]probeSpec{"m.example.com": { |
| 137 | Host: "m.example.com", |
| 138 | Addrs: []string{ts.Listener.Addr().String()}, |
| 139 | Health: provision.Health{ |
| 140 | Path: "/up", Status: 200, HostHeader: "probe.internal", IntervalMS: 20, |
| 141 | }.WithDefaults(), |
| 142 | }}) |
| 143 | waitFor(t, 2*time.Second, "the first probe with an explicit host header", func() bool { |
| 144 | h, _ := saw() |
| 145 | return h != "" |
| 146 | }) |
| 147 | if h, path := saw(); h != "probe.internal" || path != "/up" { |
| 148 | t.Fatalf("probe hit Host=%q path=%q, want probe.internal /up", h, path) |
| 149 | } |
| 150 | p.Stop() |
| 151 | |
| 152 | mu.Lock() |
| 153 | seenHost, seenPath = "", "" |
| 154 | mu.Unlock() |
| 155 | |
| 156 | // HostHeader unset: the route host is substituted — WithDefaults |
| 157 | // deliberately leaves HostHeader empty so the prober can see "unset". |
| 158 | p2 := newProberSet() |
| 159 | defer p2.Stop() |
| 160 | p2.Sync(map[string]probeSpec{"m.example.com": { |
| 161 | Host: "m.example.com", |
| 162 | Addrs: []string{ts.Listener.Addr().String()}, |
| 163 | Health: provision.Health{IntervalMS: 20}.WithDefaults(), |
| 164 | }}) |
| 165 | waitFor(t, 2*time.Second, "the first probe with a defaulted host header", func() bool { |
| 166 | h, _ := saw() |
| 167 | return h != "" |
| 168 | }) |
| 169 | if h, path := saw(); h != "m.example.com" || path != "/" { |
| 170 | t.Fatalf("probe hit Host=%q path=%q, want the route host m.example.com and /", h, path) |
| 171 | } |
| 172 | } |
| 173 | |
| 174 | func TestSyncStopsVanishedHosts(t *testing.T) { |
| 175 | backend, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| 176 | ok.Store(true) |
| 177 | |
| 178 | p := newProberSet() |
| 179 | defer p.Stop() |
| 180 | spec := probeSpec{ |
| 181 | Host: "m.example.com", |
| 182 | Addrs: []string{backend.Listener.Addr().String()}, |
| 183 | Health: provision.Health{IntervalMS: 20}.WithDefaults(), |
| 184 | } |
| 185 | p.Sync(map[string]probeSpec{"m.example.com": spec}) |
| 186 | waitFor(t, 2*time.Second, "the host to be probed at least twice", func() bool { |
| 187 | return hits.Load() >= 2 |
| 188 | }) |
| 189 | |
| 190 | // An unchanged Sync must NOT restart the prober (Reload runs on every |
| 191 | // fingerprint move; restarting would reset every counter and re-probe |
| 192 | // the fleet from scratch each time an unrelated route changed). |
| 193 | before := p.State()["m.example.com"][0].Since |
| 194 | p.Sync(map[string]probeSpec{"m.example.com": spec}) |
| 195 | if after := p.State()["m.example.com"][0].Since; !after.Equal(before) { |
| 196 | t.Errorf("unchanged Sync restarted the prober (Since %v -> %v)", before, after) |
| 197 | } |
| 198 | |
| 199 | // The host vanishes: its goroutine is stopped before Sync returns, so |
| 200 | // the hit count is frozen from here on — no settle window needed. |
| 201 | p.Sync(map[string]probeSpec{}) |
| 202 | if st := p.State(); len(st) != 0 { |
| 203 | t.Fatalf("State() = %v, want empty after the host vanished", st) |
| 204 | } |
| 205 | if live := p.Live("m.example.com"); len(live) != 0 { |
| 206 | t.Fatalf("Live() = %v, want empty after the host vanished", live) |
| 207 | } |
| 208 | frozen := hits.Load() |
| 209 | time.Sleep(150 * time.Millisecond) // >> the 20ms interval it was probing at |
| 210 | if now := hits.Load(); now != frozen { |
| 211 | t.Fatalf("backend hit %d more time(s) after Sync dropped the host — goroutine leaked", now-frozen) |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // TestSyncReplacesChangedSpecWithoutAGap covers the replace path — the one |
| 216 | // Sync branch where a host is removed AND re-added in the same call. Two |
| 217 | // invariants ride on how that is ordered, and both are here so a refactor that |
| 218 | // simplifies the ordering fails loudly: |
| 219 | // |
| 220 | // 1. Live(host) is NEVER empty across the swap. The replacement is built |
| 221 | // (all-live, per the optimism rule) inside the same critical section that |
| 222 | // removes the outgoing prober, so a request racing a Reload sees the old |
| 223 | // table or the new one, never a gap. A gap would be a 503 for a route whose |
| 224 | // upstreams are perfectly healthy — a self-inflicted outage on every |
| 225 | // addr-set edit. |
| 226 | // 2. The dropped backend is not probed again once Sync RETURNS. Retired |
| 227 | // probers are stop()ed (which waits for the goroutine to exit) before any |
| 228 | // replacement goroutine is launched, so there is no straggler tick against |
| 229 | // an upstream the route no longer names. |
| 230 | func TestSyncReplacesChangedSpecWithoutAGap(t *testing.T) { |
| 231 | const host = "m.example.com" |
| 232 | newOK := func() (*httptest.Server, *atomic.Int64) { |
| 233 | srv, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| 234 | ok.Store(true) |
| 235 | return srv, hits |
| 236 | } |
| 237 | a, _ := newOK() |
| 238 | b, bHits := newOK() |
| 239 | c, _ := newOK() |
| 240 | aAddr, bAddr, cAddr := a.Listener.Addr().String(), b.Listener.Addr().String(), c.Listener.Addr().String() |
| 241 | |
| 242 | spec := func(addrs ...string) probeSpec { |
| 243 | return probeSpec{ |
| 244 | Host: host, |
| 245 | Addrs: addrs, |
| 246 | Health: provision.Health{ |
| 247 | Path: "/", Status: 200, IntervalMS: 20, FailAfter: 3, RiseAfter: 2, |
| 248 | }.WithDefaults(), |
| 249 | } |
| 250 | } |
| 251 | |
| 252 | p := newProberSet() |
| 253 | defer p.Stop() |
| 254 | p.Sync(map[string]probeSpec{host: spec(aAddr, bAddr)}) |
| 255 | waitFor(t, 2*time.Second, "the outgoing addr set to be probed", func() bool { |
| 256 | return bHits.Load() >= 2 |
| 257 | }) |
| 258 | |
| 259 | // Hammer Live from a second goroutine for the whole duration of the Sync. |
| 260 | // No sleep in the loop: the window being guarded is a few instructions |
| 261 | // wide, so the poller has to be as tight as the race it is hunting. The |
| 262 | // polling channel is closed after its first reading, so the swap cannot |
| 263 | // start before the poller is actually running — otherwise a scheduler that |
| 264 | // gets round to the goroutine late turns the guard into a no-op. |
| 265 | stop := make(chan struct{}) |
| 266 | done := make(chan struct{}) |
| 267 | var sawGap bool |
| 268 | polls := new(atomic.Int64) |
| 269 | go func() { |
| 270 | defer close(done) |
| 271 | for { |
| 272 | polls.Add(1) |
| 273 | if live := p.Live(host); len(live) == 0 { |
| 274 | sawGap = true |
| 275 | return |
| 276 | } |
| 277 | select { |
| 278 | case <-stop: |
| 279 | return |
| 280 | default: |
| 281 | } |
| 282 | } |
| 283 | }() |
| 284 | // Do not swap until the poller has been scheduled and taken several |
| 285 | // readings — otherwise a late-starting goroutine would see only the |
| 286 | // finished state and the guard would silently assert nothing. |
| 287 | waitFor(t, 2*time.Second, "the Live poller to be running concurrently", func() bool { |
| 288 | return polls.Load() > 2 |
| 289 | }) |
| 290 | |
| 291 | p.Sync(map[string]probeSpec{host: spec(aAddr, cAddr)}) |
| 292 | |
| 293 | // Freeze the retired backend's counter the instant Sync returns: stop() is |
| 294 | // synchronous, so no settle window is allowed here. |
| 295 | frozen := bHits.Load() |
| 296 | close(stop) |
| 297 | <-done |
| 298 | if sawGap { |
| 299 | t.Fatalf("Live(%s) was empty during the spec swap — a healthy route would have 503ed", host) |
| 300 | } |
| 301 | |
| 302 | time.Sleep(150 * time.Millisecond) // >> the 20ms interval it was probing at |
| 303 | if now := bHits.Load(); now != frozen { |
| 304 | t.Fatalf("dropped upstream probed %d more time(s) after Sync returned — straggler tick from a retired prober", now-frozen) |
| 305 | } |
| 306 | |
| 307 | // The new spec is the one running: the replacement's addrs, all live. |
| 308 | live := p.Live(host) |
| 309 | if len(live) != 2 || live[0] != aAddr || live[1] != cAddr { |
| 310 | t.Fatalf("live after the swap = %v, want the new addr set [%s %s]", live, aAddr, cAddr) |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | // TestSyncNormalizesHealthBeforeComparing pins finding #2 of the fix round: the |
| 315 | // same probing job expressed with a raw Health{} and with a WithDefaults()ed one |
| 316 | // must compare EQUAL, or a caller that takes a mixed path (the registry parse on |
| 317 | // one Reload, a defaulted spec on the next) restarts every prober on the box — |
| 318 | // resetting every fail/rise run and re-probing the whole fleet — on each Reload. |
| 319 | func TestSyncNormalizesHealthBeforeComparing(t *testing.T) { |
| 320 | const host = "m.example.com" |
| 321 | backend, ok, _ := switchableBackend(t, http.StatusInternalServerError) |
| 322 | ok.Store(true) |
| 323 | |
| 324 | p := newProberSet() |
| 325 | defer p.Stop() |
| 326 | proberFor := func() *hostProber { |
| 327 | p.mu.Lock() |
| 328 | defer p.mu.Unlock() |
| 329 | return p.hosts[host] |
| 330 | } |
| 331 | |
| 332 | raw := probeSpec{ |
| 333 | Host: host, |
| 334 | Addrs: []string{backend.Listener.Addr().String()}, |
| 335 | Health: provision.Health{IntervalMS: 20}, |
| 336 | } |
| 337 | defaulted := raw |
| 338 | defaulted.Health = raw.Health.WithDefaults() |
| 339 | if raw.Health == defaulted.Health { |
| 340 | t.Fatal("the two spellings are identical; this test is asserting nothing") |
| 341 | } |
| 342 | |
| 343 | p.Sync(map[string]probeSpec{host: raw}) |
| 344 | first := proberFor() |
| 345 | if first == nil { |
| 346 | t.Fatal("no prober after the first Sync") |
| 347 | } |
| 348 | // Identity, not just Since: the exact goroutine must survive both spellings. |
| 349 | p.Sync(map[string]probeSpec{host: defaulted}) |
| 350 | if got := proberFor(); got != first { |
| 351 | t.Error("a defaulted spec restarted the prober started from the raw one") |
| 352 | } |
| 353 | p.Sync(map[string]probeSpec{host: raw}) |
| 354 | if got := proberFor(); got != first { |
| 355 | t.Error("a raw spec restarted the prober started from the defaulted one") |
| 356 | } |
| 357 | } |
| 358 | |
| 359 | func TestEjectIsImmediateAndProbeRejoins(t *testing.T) { |
| 360 | a := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 361 | w.WriteHeader(http.StatusOK) |
| 362 | })) |
| 363 | t.Cleanup(a.Close) |
| 364 | b := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 365 | w.WriteHeader(http.StatusOK) |
| 366 | })) |
| 367 | t.Cleanup(b.Close) |
| 368 | aAddr, bAddr := a.Listener.Addr().String(), b.Listener.Addr().String() |
| 369 | |
| 370 | p := newProberSet() |
| 371 | defer p.Stop() |
| 372 | p.Sync(map[string]probeSpec{"m.example.com": { |
| 373 | Host: "m.example.com", |
| 374 | Addrs: []string{aAddr, bAddr}, |
| 375 | Health: provision.Health{ |
| 376 | Path: "/", Status: 200, IntervalMS: 20, FailAfter: 3, RiseAfter: 2, |
| 377 | }.WithDefaults(), |
| 378 | }}) |
| 379 | waitFor(t, 2*time.Second, "both upstreams live", func() bool { |
| 380 | return len(p.Live("m.example.com")) == 2 |
| 381 | }) |
| 382 | |
| 383 | // Eject is the in-request dial-failure fast path: the drop is visible on |
| 384 | // the very next Live(), with no probe tick in between. |
| 385 | p.Eject("m.example.com", aAddr) |
| 386 | live := p.Live("m.example.com") |
| 387 | if len(live) != 1 || live[0] != bAddr { |
| 388 | t.Fatalf("live right after Eject = %v, want only %s", live, bAddr) |
| 389 | } |
| 390 | got := p.State()["m.example.com"][0] |
| 391 | if got.Addr != aAddr || got.Up { |
| 392 | t.Fatalf("ejected upstream state = %+v, want %s down", got, aAddr) |
| 393 | } |
| 394 | if got.Fails != 3 { |
| 395 | t.Errorf("ejected upstream Fails = %d, want FailAfter (3)", got.Fails) |
| 396 | } |
| 397 | if got.LastError == "" { |
| 398 | t.Error("ejected upstream LastError is empty, want the ejection reason") |
| 399 | } |
| 400 | |
| 401 | // The upstream was never actually unhealthy, so probe passes rejoin it. |
| 402 | waitFor(t, 2*time.Second, "the ejected upstream to rejoin on probe passes", func() bool { |
| 403 | return len(p.Live("m.example.com")) == 2 |
| 404 | }) |
| 405 | |
| 406 | // Ejecting an addr this host does not have (or an unknown host) is a |
| 407 | // no-op, not a panic: the proxy can race a Reload that dropped either. |
| 408 | p.Eject("m.example.com", "203.0.113.9:1") |
| 409 | p.Eject("nobody.example.com", aAddr) |
| 410 | if len(p.Live("m.example.com")) != 2 { |
| 411 | t.Fatalf("stray Eject disturbed the live set: %v", p.Live("m.example.com")) |
| 412 | } |
| 413 | } |
| 414 | |
| 415 | // probeState reads one upstream's state off a hostProber the way State() would, |
| 416 | // for tests that drive record() directly instead of waiting on probe ticks. |
| 417 | func probeState(hp *hostProber, addr string) UpstreamState { |
| 418 | hp.mu.Lock() |
| 419 | defer hp.mu.Unlock() |
| 420 | st := hp.state[addr] |
| 421 | if st == nil { |
| 422 | return UpstreamState{} |
| 423 | } |
| 424 | return UpstreamState{Addr: addr, Up: st.up, Since: st.since, Fails: st.fails, LastError: st.lastError} |
| 425 | } |
| 426 | |
| 427 | // TestProberClampsNonPositiveInterval guards the clamp that keeps a bad health |
| 428 | // JSON from taking the whole edge process down: time.NewTicker PANICS on a |
| 429 | // non-positive duration, and the prober's ticker lives in the goroutine that |
| 430 | // serves every probed route on the box. Sync's WithDefaults normalization |
| 431 | // handles interval_ms 0; it passes NEGATIVES straight through, and |
| 432 | // SetUpstreams (plus the route PATCH endpoint that follows it) writes the |
| 433 | // health column directly, so the clamp is the only thing between a -1 and a |
| 434 | // crash. |
| 435 | func TestProberClampsNonPositiveInterval(t *testing.T) { |
| 436 | for _, ms := range []int{0, -1, -60000} { |
| 437 | hp := newHostProber(probeSpec{ |
| 438 | Host: "m.example.com", |
| 439 | Addrs: []string{"10.0.0.1:80"}, |
| 440 | Health: provision.Health{IntervalMS: ms}, |
| 441 | }, nil) |
| 442 | if hp.interval != fallbackProbeInterval { |
| 443 | t.Errorf("interval_ms %d clamped to %v, want the %v fallback", ms, hp.interval, fallbackProbeInterval) |
| 444 | } |
| 445 | } |
| 446 | |
| 447 | // And end to end through Sync: a negative interval survives WithDefaults, |
| 448 | // so this is the shape that would reach time.NewTicker. It must probe, not |
| 449 | // panic. |
| 450 | backend, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| 451 | ok.Store(true) |
| 452 | p := newProberSet() |
| 453 | defer p.Stop() |
| 454 | p.Sync(map[string]probeSpec{"m.example.com": { |
| 455 | Host: "m.example.com", |
| 456 | Addrs: []string{backend.Listener.Addr().String()}, |
| 457 | Health: provision.Health{Path: "/", Status: 200, IntervalMS: -1}, |
| 458 | }}) |
| 459 | waitFor(t, 2*time.Second, "the first probe of a negative-interval spec", func() bool { |
| 460 | return hits.Load() >= 1 |
| 461 | }) |
| 462 | time.Sleep(300 * time.Millisecond) |
| 463 | // At the clamped 2000ms cadence this window holds the immediate first |
| 464 | // probe and possibly a second on a slow box; an unclamped interval would |
| 465 | // be in the hundreds. A bound, not an exact count. |
| 466 | if n := hits.Load(); n > 5 { |
| 467 | t.Fatalf("backend probed %d times in ~300ms — the negative interval was not clamped to %v", n, fallbackProbeInterval) |
| 468 | } |
| 469 | } |
| 470 | |
| 471 | // TestProberClampsNonPositiveFailAndRiseCounts is the other half of the same |
| 472 | // amendment — a non-positive fail_after must not fail an upstream away on a |
| 473 | // single blip, nor a non-positive rise_after rejoin it on a single pass — and |
| 474 | // it also pins the recovery-state truth: a DOWN upstream keeps its Fails and |
| 475 | // LastError through the passing probes of a recovery, so the dashboard never |
| 476 | // shows `Up:false, Fails:0, LastError:""` (down for no stated reason) to an |
| 477 | // operator watching a failover. Driven through record() directly, so it asserts |
| 478 | // exact counts with no timing in play. |
| 479 | func TestProberClampsNonPositiveFailAndRiseCounts(t *testing.T) { |
| 480 | const addr = "10.0.0.1:80" |
| 481 | hp := newHostProber(probeSpec{ |
| 482 | Host: "m.example.com", |
| 483 | Addrs: []string{addr}, |
| 484 | Health: provision.Health{Path: "/", Status: 200, IntervalMS: 20, FailAfter: -1, RiseAfter: 0}, |
| 485 | }, nil) |
| 486 | if hp.failAt != fallbackFailAfter || hp.riseAt != fallbackRiseAfter { |
| 487 | t.Fatalf("clamped counts = fail %d rise %d, want %d/%d", hp.failAt, hp.riseAt, fallbackFailAfter, fallbackRiseAfter) |
| 488 | } |
| 489 | |
| 490 | boom := errors.New("dial: connection refused") |
| 491 | for i := 1; i < fallbackFailAfter; i++ { |
| 492 | hp.record(addr, boom) |
| 493 | if st := probeState(hp, addr); !st.Up { |
| 494 | t.Fatalf("upstream went down after %d failure(s), want the clamped fail_after (%d)", i, fallbackFailAfter) |
| 495 | } |
| 496 | } |
| 497 | hp.record(addr, boom) |
| 498 | down := probeState(hp, addr) |
| 499 | if down.Up { |
| 500 | t.Fatalf("upstream still up after %d failures, want down", fallbackFailAfter) |
| 501 | } |
| 502 | if down.Fails != fallbackFailAfter || down.LastError != boom.Error() { |
| 503 | t.Fatalf("down state = %+v, want %d fails and the dial reason", down, fallbackFailAfter) |
| 504 | } |
| 505 | |
| 506 | for i := 1; i < fallbackRiseAfter; i++ { |
| 507 | hp.record(addr, nil) |
| 508 | st := probeState(hp, addr) |
| 509 | if st.Up { |
| 510 | t.Fatalf("upstream rejoined after %d pass(es), want the clamped rise_after (%d)", i, fallbackRiseAfter) |
| 511 | } |
| 512 | // Still down → still owes the operator a reason. |
| 513 | if st.Fails != fallbackFailAfter || st.LastError != boom.Error() { |
| 514 | t.Fatalf("recovering (still down) state = %+v, want the failure count and reason held until it comes up", st) |
| 515 | } |
| 516 | } |
| 517 | hp.record(addr, nil) |
| 518 | up := probeState(hp, addr) |
| 519 | if !up.Up || up.Fails != 0 || up.LastError != "" { |
| 520 | t.Fatalf("recovered state = %+v, want up with fails and error cleared on the transition", up) |
| 521 | } |
| 522 | } |
| 523 | |
| 524 | // TestProberStateReturnsCopies pins the Task 6 contract: the status snapshot |
| 525 | // serializes what State() returns while probes keep running, so the snapshot |
| 526 | // must never share a slice with the live table. |
| 527 | func TestProberStateReturnsCopies(t *testing.T) { |
| 528 | backend, ok, _ := switchableBackend(t, http.StatusInternalServerError) |
| 529 | ok.Store(true) |
| 530 | |
| 531 | p := newProberSet() |
| 532 | defer p.Stop() |
| 533 | addr := backend.Listener.Addr().String() |
| 534 | p.Sync(map[string]probeSpec{"m.example.com": { |
| 535 | Host: "m.example.com", |
| 536 | Addrs: []string{addr}, |
| 537 | Health: provision.Health{IntervalMS: 20}.WithDefaults(), |
| 538 | }}) |
| 539 | |
| 540 | st := p.State() |
| 541 | st["m.example.com"][0].Addr = "clobbered" |
| 542 | delete(st, "m.example.com") |
| 543 | if again := p.State()["m.example.com"]; len(again) != 1 || again[0].Addr != addr { |
| 544 | t.Fatalf("State() = %v after a caller mutated its copy, want the live table intact", again) |
| 545 | } |
| 546 | live := p.Live("m.example.com") |
| 547 | if len(live) != 1 { |
| 548 | t.Fatalf("Live() = %v, want the single upstream", live) |
| 549 | } |
| 550 | live[0] = "clobbered" |
| 551 | if again := p.Live("m.example.com"); again[0] != addr { |
| 552 | t.Fatalf("Live() = %v after a caller mutated its copy, want %s", again, addr) |
| 553 | } |
| 554 | } |
| 555 | |
amadan