| index 962ea780b..381e99e67 100644 |
| --- a/internal/edge/prober_test.go |
| +++ b/internal/edge/prober_test.go |
| @@ -1,6 +1,7 @@ |
| package edge |
| |
| import ( |
| + "errors" |
| "net/http" |
| "net/http/httptest" |
| "sync" |
| @@ -211,6 +212,150 @@ func TestSyncStopsVanishedHosts(t *testing.T) { |
| } |
| } |
| |
| +// TestSyncReplacesChangedSpecWithoutAGap covers the replace path — the one |
| +// Sync branch where a host is removed AND re-added in the same call. Two |
| +// invariants ride on how that is ordered, and both are here so a refactor that |
| +// simplifies the ordering fails loudly: |
| +// |
| +// 1. Live(host) is NEVER empty across the swap. The replacement is built |
| +// (all-live, per the optimism rule) inside the same critical section that |
| +// removes the outgoing prober, so a request racing a Reload sees the old |
| +// table or the new one, never a gap. A gap would be a 503 for a route whose |
| +// upstreams are perfectly healthy — a self-inflicted outage on every |
| +// addr-set edit. |
| +// 2. The dropped backend is not probed again once Sync RETURNS. Retired |
| +// probers are stop()ed (which waits for the goroutine to exit) before any |
| +// replacement goroutine is launched, so there is no straggler tick against |
| +// an upstream the route no longer names. |
| +func TestSyncReplacesChangedSpecWithoutAGap(t *testing.T) { |
| + const host = "m.example.com" |
| + newOK := func() (*httptest.Server, *atomic.Int64) { |
| + srv, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| + ok.Store(true) |
| + return srv, hits |
| + } |
| + a, _ := newOK() |
| + b, bHits := newOK() |
| + c, _ := newOK() |
| + aAddr, bAddr, cAddr := a.Listener.Addr().String(), b.Listener.Addr().String(), c.Listener.Addr().String() |
| + |
| + spec := func(addrs ...string) probeSpec { |
| + return probeSpec{ |
| + Host: host, |
| + Addrs: addrs, |
| + Health: provision.Health{ |
| + Path: "/", Status: 200, IntervalMS: 20, FailAfter: 3, RiseAfter: 2, |
| + }.WithDefaults(), |
| + } |
| + } |
| + |
| + p := newProberSet() |
| + defer p.Stop() |
| + p.Sync(map[string]probeSpec{host: spec(aAddr, bAddr)}) |
| + waitFor(t, 2*time.Second, "the outgoing addr set to be probed", func() bool { |
| + return bHits.Load() >= 2 |
| + }) |
| + |
| + // Hammer Live from a second goroutine for the whole duration of the Sync. |
| + // No sleep in the loop: the window being guarded is a few instructions |
| + // wide, so the poller has to be as tight as the race it is hunting. The |
| + // polling channel is closed after its first reading, so the swap cannot |
| + // start before the poller is actually running — otherwise a scheduler that |
| + // gets round to the goroutine late turns the guard into a no-op. |
| + stop := make(chan struct{}) |
| + done := make(chan struct{}) |
| + var sawGap bool |
| + polls := new(atomic.Int64) |
| + go func() { |
| + defer close(done) |
| + for { |
| + polls.Add(1) |
| + if live := p.Live(host); len(live) == 0 { |
| + sawGap = true |
| + return |
| + } |
| + select { |
| + case <-stop: |
| + return |
| + default: |
| + } |
| + } |
| + }() |
| + // Do not swap until the poller has been scheduled and taken several |
| + // readings — otherwise a late-starting goroutine would see only the |
| + // finished state and the guard would silently assert nothing. |
| + waitFor(t, 2*time.Second, "the Live poller to be running concurrently", func() bool { |
| + return polls.Load() > 2 |
| + }) |
| + |
| + p.Sync(map[string]probeSpec{host: spec(aAddr, cAddr)}) |
| + |
| + // Freeze the retired backend's counter the instant Sync returns: stop() is |
| + // synchronous, so no settle window is allowed here. |
| + frozen := bHits.Load() |
| + close(stop) |
| + <-done |
| + if sawGap { |
| + t.Fatalf("Live(%s) was empty during the spec swap — a healthy route would have 503ed", host) |
| + } |
| + |
| + time.Sleep(150 * time.Millisecond) // >> the 20ms interval it was probing at |
| + if now := bHits.Load(); now != frozen { |
| + t.Fatalf("dropped upstream probed %d more time(s) after Sync returned — straggler tick from a retired prober", now-frozen) |
| + } |
| + |
| + // The new spec is the one running: the replacement's addrs, all live. |
| + live := p.Live(host) |
| + if len(live) != 2 || live[0] != aAddr || live[1] != cAddr { |
| + t.Fatalf("live after the swap = %v, want the new addr set [%s %s]", live, aAddr, cAddr) |
| + } |
| +} |
| + |
| +// TestSyncNormalizesHealthBeforeComparing pins finding #2 of the fix round: the |
| +// same probing job expressed with a raw Health{} and with a WithDefaults()ed one |
| +// must compare EQUAL, or a caller that takes a mixed path (the registry parse on |
| +// one Reload, a defaulted spec on the next) restarts every prober on the box — |
| +// resetting every fail/rise run and re-probing the whole fleet — on each Reload. |
| +func TestSyncNormalizesHealthBeforeComparing(t *testing.T) { |
| + const host = "m.example.com" |
| + backend, ok, _ := switchableBackend(t, http.StatusInternalServerError) |
| + ok.Store(true) |
| + |
| + p := newProberSet() |
| + defer p.Stop() |
| + proberFor := func() *hostProber { |
| + p.mu.Lock() |
| + defer p.mu.Unlock() |
| + return p.hosts[host] |
| + } |
| + |
| + raw := probeSpec{ |
| + Host: host, |
| + Addrs: []string{backend.Listener.Addr().String()}, |
| + Health: provision.Health{IntervalMS: 20}, |
| + } |
| + defaulted := raw |
| + defaulted.Health = raw.Health.WithDefaults() |
| + if raw.Health == defaulted.Health { |
| + t.Fatal("the two spellings are identical; this test is asserting nothing") |
| + } |
| + |
| + p.Sync(map[string]probeSpec{host: raw}) |
| + first := proberFor() |
| + if first == nil { |
| + t.Fatal("no prober after the first Sync") |
| + } |
| + // Identity, not just Since: the exact goroutine must survive both spellings. |
| + p.Sync(map[string]probeSpec{host: defaulted}) |
| + if got := proberFor(); got != first { |
| + t.Error("a defaulted spec restarted the prober started from the raw one") |
| + } |
| + p.Sync(map[string]probeSpec{host: raw}) |
| + if got := proberFor(); got != first { |
| + t.Error("a raw spec restarted the prober started from the defaulted one") |
| + } |
| +} |
| + |
| func TestEjectIsImmediateAndProbeRejoins(t *testing.T) { |
| a := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| w.WriteHeader(http.StatusOK) |
| @@ -267,66 +412,112 @@ func TestEjectIsImmediateAndProbeRejoins(t *testing.T) { |
| } |
| } |
| |
| -// TestProberClampsZeroIntervalToTheDefault is the defensive half of the |
| -// controller amendment: mint-time defaulting (WithDefaults) is NOT the only |
| -// writer of a route's health JSON — SetUpstreams (and the PATCH endpoint that |
| -// follows it) can write the object directly — so a spec that reaches the |
| -// prober with interval_ms 0 must be clamped here, not tight-loop against a |
| -// customer's upstream at whatever rate the network allows. |
| -func TestProberClampsZeroIntervalToTheDefault(t *testing.T) { |
| +// probeState reads one upstream's state off a hostProber the way State() would, |
| +// for tests that drive record() directly instead of waiting on probe ticks. |
| +func probeState(hp *hostProber, addr string) UpstreamState { |
| + hp.mu.Lock() |
| + defer hp.mu.Unlock() |
| + st := hp.state[addr] |
| + if st == nil { |
| + return UpstreamState{} |
| + } |
| + return UpstreamState{Addr: addr, Up: st.up, Since: st.since, Fails: st.fails, LastError: st.lastError} |
| +} |
| + |
| +// TestProberClampsNonPositiveInterval guards the clamp that keeps a bad health |
| +// JSON from taking the whole edge process down: time.NewTicker PANICS on a |
| +// non-positive duration, and the prober's ticker lives in the goroutine that |
| +// serves every probed route on the box. Sync's WithDefaults normalization |
| +// handles interval_ms 0; it passes NEGATIVES straight through, and |
| +// SetUpstreams (plus the route PATCH endpoint that follows it) writes the |
| +// health column directly, so the clamp is the only thing between a -1 and a |
| +// crash. |
| +func TestProberClampsNonPositiveInterval(t *testing.T) { |
| + for _, ms := range []int{0, -1, -60000} { |
| + hp := newHostProber(probeSpec{ |
| + Host: "m.example.com", |
| + Addrs: []string{"10.0.0.1:80"}, |
| + Health: provision.Health{IntervalMS: ms}, |
| + }, nil) |
| + if hp.interval != fallbackProbeInterval { |
| + t.Errorf("interval_ms %d clamped to %v, want the %v fallback", ms, hp.interval, fallbackProbeInterval) |
| + } |
| + } |
| + |
| + // And end to end through Sync: a negative interval survives WithDefaults, |
| + // so this is the shape that would reach time.NewTicker. It must probe, not |
| + // panic. |
| backend, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| ok.Store(true) |
| - |
| p := newProberSet() |
| defer p.Stop() |
| - // Deliberately NOT WithDefaults()ed — this is the un-defaulted shape a |
| - // direct health-JSON writer can produce. |
| p.Sync(map[string]probeSpec{"m.example.com": { |
| Host: "m.example.com", |
| Addrs: []string{backend.Listener.Addr().String()}, |
| - Health: provision.Health{Path: "/", Status: 200}, |
| + Health: provision.Health{Path: "/", Status: 200, IntervalMS: -1}, |
| }}) |
| - waitFor(t, 2*time.Second, "the first probe", func() bool { return hits.Load() >= 1 }) |
| - |
| - time.Sleep(400 * time.Millisecond) |
| - // At the clamped 2000ms cadence this window holds one probe (the |
| - // immediate first one) and possibly a second on a slow box. An unclamped |
| - // zero interval would be in the hundreds or thousands. |
| + waitFor(t, 2*time.Second, "the first probe of a negative-interval spec", func() bool { |
| + return hits.Load() >= 1 |
| + }) |
| + time.Sleep(300 * time.Millisecond) |
| + // At the clamped 2000ms cadence this window holds the immediate first |
| + // probe and possibly a second on a slow box; an unclamped interval would |
| + // be in the hundreds. A bound, not an exact count. |
| if n := hits.Load(); n > 5 { |
| - t.Fatalf("backend probed %d times in ~400ms — zero interval was not clamped to 2000ms", n) |
| + t.Fatalf("backend probed %d times in ~300ms — the negative interval was not clamped to %v", n, fallbackProbeInterval) |
| } |
| } |
| |
| -// TestProberClampsZeroFailAndRiseCounts is the other half of the same |
| -// amendment: a zero fail_after must not fail an upstream away on a single |
| -// blip, and a zero rise_after must not rejoin it on a single pass. |
| -func TestProberClampsZeroFailAndRiseCounts(t *testing.T) { |
| - backend, ok, hits := switchableBackend(t, http.StatusInternalServerError) |
| - |
| - p := newProberSet() |
| - defer p.Stop() |
| - p.Sync(map[string]probeSpec{"m.example.com": { |
| - Host: "m.example.com", |
| - Addrs: []string{backend.Listener.Addr().String()}, |
| - // Interval is set (the clamp for it has its own test); the counts |
| - // are the zeroes under test. |
| - Health: provision.Health{Path: "/", Status: 200, IntervalMS: 20}, |
| - }}) |
| +// TestProberClampsNonPositiveFailAndRiseCounts is the other half of the same |
| +// amendment — a non-positive fail_after must not fail an upstream away on a |
| +// single blip, nor a non-positive rise_after rejoin it on a single pass — and |
| +// it also pins the recovery-state truth: a DOWN upstream keeps its Fails and |
| +// LastError through the passing probes of a recovery, so the dashboard never |
| +// shows `Up:false, Fails:0, LastError:""` (down for no stated reason) to an |
| +// operator watching a failover. Driven through record() directly, so it asserts |
| +// exact counts with no timing in play. |
| +func TestProberClampsNonPositiveFailAndRiseCounts(t *testing.T) { |
| + const addr = "10.0.0.1:80" |
| + hp := newHostProber(probeSpec{ |
| + Host: "m.example.com", |
| + Addrs: []string{addr}, |
| + Health: provision.Health{Path: "/", Status: 200, IntervalMS: 20, FailAfter: -1, RiseAfter: 0}, |
| + }, nil) |
| + if hp.failAt != fallbackFailAfter || hp.riseAt != fallbackRiseAfter { |
| + t.Fatalf("clamped counts = fail %d rise %d, want %d/%d", hp.failAt, hp.riseAt, fallbackFailAfter, fallbackRiseAfter) |
| + } |
| |
| - waitFor(t, 2*time.Second, "the failing upstream to go down", func() bool { |
| - return len(p.Live("m.example.com")) == 0 |
| - }) |
| - atDown := hits.Load() |
| - if atDown < 3 { |
| - t.Fatalf("upstream went down after %d probe(s), want >= FailAfter's clamped 3", atDown) |
| + boom := errors.New("dial: connection refused") |
| + for i := 1; i < fallbackFailAfter; i++ { |
| + hp.record(addr, boom) |
| + if st := probeState(hp, addr); !st.Up { |
| + t.Fatalf("upstream went down after %d failure(s), want the clamped fail_after (%d)", i, fallbackFailAfter) |
| + } |
| + } |
| + hp.record(addr, boom) |
| + down := probeState(hp, addr) |
| + if down.Up { |
| + t.Fatalf("upstream still up after %d failures, want down", fallbackFailAfter) |
| + } |
| + if down.Fails != fallbackFailAfter || down.LastError != boom.Error() { |
| + t.Fatalf("down state = %+v, want %d fails and the dial reason", down, fallbackFailAfter) |
| } |
| |
| - ok.Store(true) |
| - waitFor(t, 2*time.Second, "the recovered upstream to rejoin", func() bool { |
| - return len(p.Live("m.example.com")) == 1 |
| - }) |
| - if n := hits.Load(); n < atDown+2 { |
| - t.Fatalf("upstream rejoined after %d pass(es), want >= RiseAfter's clamped 2", n-atDown) |
| + for i := 1; i < fallbackRiseAfter; i++ { |
| + hp.record(addr, nil) |
| + st := probeState(hp, addr) |
| + if st.Up { |
| + t.Fatalf("upstream rejoined after %d pass(es), want the clamped rise_after (%d)", i, fallbackRiseAfter) |
| + } |
| + // Still down → still owes the operator a reason. |
| + if st.Fails != fallbackFailAfter || st.LastError != boom.Error() { |
| + t.Fatalf("recovering (still down) state = %+v, want the failure count and reason held until it comes up", st) |
| + } |
| + } |
| + hp.record(addr, nil) |
| + up := probeState(hp, addr) |
| + if !up.Up || up.Fails != 0 || up.LastError != "" { |
| + t.Fatalf("recovered state = %+v, want up with fails and error cleared on the transition", up) |
| } |
| } |
| |
amadan