| new file mode 100644 |
| index 000000000..b6b058648 |
| --- /dev/null |
| +++ b/internal/edge/prober.go |
| @@ -0,0 +1,495 @@ |
| +package edge |
| + |
| +// prober.go — active health probing for multi-upstream addr routes (HA design |
| +// Part II, issue #152). One goroutine per probed host GETs each of its |
| +// upstreams on a fixed cadence and maintains the LIVE SET the request path |
| +// (multiProxy) picks from; the same table is the dashboard's per-upstream |
| +// state. |
| +// |
| +// Two rules shape everything below: |
| +// |
| +// 1. **Optimistic start.** Every upstream of a newly synced host starts LIVE. |
| +// A reload happens on any registry fingerprint move — an unrelated route |
| +// changing must never take a healthy fleet off the air for the first |
| +// fail_after × interval while probes catch up. Probing only ever REMOVES |
| +// confidence; it never withholds it. |
| +// |
| +// 2. **Probes dial upstreams directly**, never the edge's own public host. |
| +// Combined with the no-hibernate invariant for addr routes (Reload's `hib` |
| +// gate), a probe can therefore never wake a hibernating tenant — the wake |
| +// hazard that would otherwise make active probing incompatible with |
| +// hibernation simply does not exist here. |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "io" |
| + "log" |
| + "net/http" |
| + "slices" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// probeTimeout bounds a single probe request end to end (connect through |
| +// body). Held at 2s per the spec: long enough that a loaded-but-alive |
| +// upstream still answers, short enough that a black-holed one cannot stall a |
| +// tick past the default 2s interval. |
| +const probeTimeout = 2 * time.Second |
| + |
| +// Defensive fallbacks for a spec that reaches the prober with a non-positive |
| +// cadence or count. These duplicate provision.Health.WithDefaults on purpose: |
| +// WithDefaults runs at MINT time, and mint is not the only writer of a route's |
| +// health JSON — registry.SetUpstreams writes the column directly, and the |
| +// route PATCH endpoint will too. A zero interval arriving from any of those |
| +// must cost one 2s-cadence prober, not an unbounded tight loop against a |
| +// customer's upstream; a zero fail_after must not fail an upstream away on a |
| +// single blip, nor a zero rise_after rejoin it on a single pass. |
| +// |
| +// Only NON-POSITIVE values are replaced. A deliberately fast interval is the |
| +// operator's call (and the tests' — they probe at 20ms). |
| +const ( |
| + fallbackProbeInterval = 2000 * time.Millisecond |
| + fallbackFailAfter = 3 |
| + fallbackRiseAfter = 2 |
| +) |
| + |
| +// UpstreamState is the dashboard-facing view of one upstream: what the probe |
| +// currently believes, when that belief last changed, and why. Task 6 carries |
| +// this into the box status snapshot, so the JSON tags are wire shape. |
| +type UpstreamState struct { |
| + Addr string `json:"addr"` |
| + Up bool `json:"up"` |
| + Since time.Time `json:"since"` // last transition (or the sync that started this prober) |
| + Fails int `json:"fails"` // consecutive probe failures |
| + LastError string `json:"last_error,omitempty"` // dial vs status, for the log/dashboard |
| +} |
| + |
| +// probeSpec is one route's probing contract, parsed from the registry's health |
| +// JSON once per Reload. Health is expected (but not required) to have been |
| +// through WithDefaults — see the clamps above for why "not required". |
| +type probeSpec struct { |
| + Host string // route host — the probe's Host header default |
| + Addrs []string |
| + Health provision.Health |
| +} |
| + |
| +// equal reports whether two specs describe the same probing job. Sync uses it |
| +// to decide restart-or-leave-alone, so it must cover every field the running |
| +// goroutine captured: host (the default Host header), the addr list INCLUDING |
| +// order (round-robin walks it), and the whole health object. |
| +func (s probeSpec) equal(o probeSpec) bool { |
| + return s.Host == o.Host && s.Health == o.Health && slices.Equal(s.Addrs, o.Addrs) |
| +} |
| + |
| +// proberSet owns one prober goroutine per probed host plus the live-state |
| +// table both the request path and the status snapshot read. |
| +// |
| +// Lifecycle mirrors Activator.RetireExcept: Reload computes the desired specs |
| +// and calls Sync, which starts probers for new hosts, stops those whose route |
| +// vanished, and replaces those whose spec changed. It is the only other thing |
| +// in this package that owns per-route goroutines across a reload. |
| +// |
| +// A *proberSet must come from newProberSet; the zero value is not usable and a |
| +// nil receiver panics rather than quietly reporting "no live upstreams" — a |
| +// silent nil would turn a wiring mistake into fleet-wide 503s instead of a |
| +// loud failure in the first test that hits it. |
| +type proberSet struct { |
| + // mu guards hosts only. The per-host state table lives on hostProber |
| + // under its own lock, so a probe tick writing state never contends with |
| + // Sync, and Sync never has to hold this lock while waiting for a |
| + // goroutine to exit (which would stall Live on the request hot path). |
| + mu sync.Mutex |
| + hosts map[string]*hostProber |
| + client *http.Client |
| +} |
| + |
| +func newProberSet() *proberSet { |
| + return &proberSet{ |
| + hosts: make(map[string]*hostProber), |
| + client: &http.Client{ |
| + Timeout: probeTimeout, |
| + // A redirect is not health: an upstream answering 302 has not |
| + // served the configured status, and following the hop would |
| + // probe something the operator never named. |
| + CheckRedirect: func(*http.Request, []*http.Request) error { |
| + return http.ErrUseLastResponse |
| + }, |
| + Transport: &http.Transport{ |
| + // Probes are the same tiny request to the same handful of |
| + // addrs forever, so idle connections are worth keeping — |
| + // bounded tightly, since every probed upstream holds one. |
| + MaxIdleConnsPerHost: 1, |
| + IdleConnTimeout: 90 * time.Second, |
| + ResponseHeaderTimeout: probeTimeout, |
| + }, |
| + }, |
| + } |
| +} |
| + |
| +// hostProber is one host's probing job: its spec, the clamped cadence derived |
| +// from it, its goroutine's stop handle, and the per-addr state that goroutine |
| +// maintains. |
| +// |
| +// The state table is OWNED BY THIS STRUCT, not by the set — that is what makes |
| +// Sync's replace safe. A replaced prober's goroutine can still be mid-probe |
| +// when the new one is installed, and whatever it writes lands in its own |
| +// retired table, which nothing reads any more. (Sync additionally waits for |
| +// the old goroutine to observe its stop before starting the new one, so even |
| +// the log lines stay ordered.) |
| +type hostProber struct { |
| + spec probeSpec |
| + client *http.Client |
| + path string |
| + wantCode int |
| + hostHdr string |
| + interval time.Duration |
| + failAt int |
| + riseAt int |
| + |
| + cancel context.CancelFunc |
| + ctx context.Context |
| + done chan struct{} |
| + |
| + mu sync.Mutex |
| + order []string // addrs in spec order — Live/State iterate this, not the map |
| + state map[string]*upstreamState |
| +} |
| + |
| +// upstreamState is the mutable half of UpstreamState plus the pass counter, |
| +// which is bookkeeping the dashboard has no use for. |
| +type upstreamState struct { |
| + up bool |
| + since time.Time |
| + fails int |
| + passes int |
| + lastError string |
| +} |
| + |
| +func newHostProber(spec probeSpec, client *http.Client) *hostProber { |
| + h := spec.Health |
| + hp := &hostProber{ |
| + spec: spec, |
| + client: client, |
| + path: h.Path, |
| + wantCode: h.Status, |
| + hostHdr: h.HostHeader, |
| + interval: time.Duration(h.IntervalMS) * time.Millisecond, |
| + failAt: h.FailAfter, |
| + riseAt: h.RiseAfter, |
| + done: make(chan struct{}), |
| + state: make(map[string]*upstreamState, len(spec.Addrs)), |
| + } |
| + if hp.path == "" { |
| + hp.path = "/" |
| + } |
| + if hp.path[0] != '/' { |
| + hp.path = "/" + hp.path |
| + } |
| + if hp.wantCode == 0 { |
| + hp.wantCode = http.StatusOK |
| + } |
| + if hp.hostHdr == "" { |
| + // WithDefaults deliberately leaves HostHeader empty so this |
| + // substitution can see "unset" rather than a value it invented. |
| + hp.hostHdr = spec.Host |
| + } |
| + if h.IntervalMS <= 0 { |
| + hp.interval = fallbackProbeInterval |
| + } |
| + if hp.failAt <= 0 { |
| + hp.failAt = fallbackFailAfter |
| + } |
| + if hp.riseAt <= 0 { |
| + hp.riseAt = fallbackRiseAfter |
| + } |
| + now := time.Now() |
| + for _, addr := range spec.Addrs { |
| + if addr == "" || hp.state[addr] != nil { |
| + continue // validation refuses both; be inert if one slips through |
| + } |
| + hp.order = append(hp.order, addr) |
| + // Rule 1: everything starts live. |
| + hp.state[addr] = &upstreamState{up: true, since: now} |
| + } |
| + hp.ctx, hp.cancel = context.WithCancel(context.Background()) |
| + return hp |
| +} |
| + |
| +// stop cancels the goroutine's context — aborting any in-flight probe at once, |
| +// so this does not wait out the 2s client timeout — and blocks until it has |
| +// exited. |
| +func (hp *hostProber) stop() { |
| + hp.cancel() |
| + <-hp.done |
| +} |
| + |
| +// Sync starts, stops and replaces probers so the running set matches want. |
| +// Called from Reload after the upstream map swap, beside RetireExcept. |
| +// |
| +// Specs with no addrs are not probed (nothing to probe); such a host holds no |
| +// entry, and Live/State report it as absent. |
| +func (p *proberSet) Sync(want map[string]probeSpec) { |
| + var retired, started []*hostProber |
| + |
| + p.mu.Lock() |
| + for host, hp := range p.hosts { |
| + w, ok := want[host] |
| + if ok && len(w.Addrs) > 0 && hp.spec.equal(w) { |
| + continue // unchanged: leave the goroutine and its counters alone |
| + } |
| + retired = append(retired, hp) |
| + delete(p.hosts, host) |
| + } |
| + for host, w := range want { |
| + if len(w.Addrs) == 0 { |
| + continue |
| + } |
| + if _, ok := p.hosts[host]; ok { |
| + continue |
| + } |
| + // Built (and therefore all-live) INSIDE the same critical section |
| + // that removed the outgoing prober, so a concurrent Live never |
| + // observes a replaced host as having no upstreams — a gap there |
| + // would be a 503 for a route that is perfectly healthy. |
| + hp := newHostProber(w, p.client) |
| + p.hosts[host] = hp |
| + started = append(started, hp) |
| + } |
| + p.mu.Unlock() |
| + |
| + // Outside the lock, and before the replacements run: retiring waits for |
| + // each old goroutine to exit, and blocking Live (request hot path) for |
| + // that wait would be exactly the stall this whole feature exists to |
| + // avoid. |
| + for _, hp := range retired { |
| + hp.stop() |
| + } |
| + for _, hp := range started { |
| + go hp.run() |
| + } |
| +} |
| + |
| +// Live reports the currently-live upstreams for host, in the route's addr |
| +// order, as a fresh slice the caller may keep. Unknown host (or one with no |
| +// live upstream) → empty: the request path answers 503 + Retry-After rather |
| +// than dialing something the probe says is dead. |
| +func (p *proberSet) Live(host string) []string { |
| + p.mu.Lock() |
| + hp := p.hosts[host] |
| + p.mu.Unlock() |
| + if hp == nil { |
| + return nil |
| + } |
| + hp.mu.Lock() |
| + defer hp.mu.Unlock() |
| + out := make([]string, 0, len(hp.order)) |
| + for _, addr := range hp.order { |
| + if hp.state[addr].up { |
| + out = append(out, addr) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// Eject drops addr from host's live set immediately — the in-request |
| +// dial-failure fast path, so a box that died between probe ticks costs one |
| +// retry rather than up to fail_after × interval of 502s. Fails is set to |
| +// failAfter so the upstream is exactly as down as a probe-failed one; probe |
| +// passes rejoin it after rise_after, the same as any other recovery. |
| +// |
| +// A host or addr this set does not know is a no-op: the request path can |
| +// legitimately race a Reload that just dropped either. |
| +func (p *proberSet) Eject(host, addr string) { |
| + p.mu.Lock() |
| + hp := p.hosts[host] |
| + p.mu.Unlock() |
| + if hp == nil { |
| + return |
| + } |
| + hp.mu.Lock() |
| + st := hp.state[addr] |
| + if st == nil { |
| + hp.mu.Unlock() |
| + return |
| + } |
| + const reason = "dial failed on a live request" |
| + was := st.up |
| + st.up = false |
| + st.passes = 0 |
| + st.fails = hp.failAt |
| + st.lastError = reason |
| + if was { |
| + st.since = time.Now() |
| + } |
| + hp.mu.Unlock() |
| + if was { |
| + log.Printf("edge: probe %s %s down (%s)", host, addr, reason) |
| + } |
| +} |
| + |
| +// State returns the probe's view of every probed host, keyed by host, each |
| +// upstream in the route's addr order. Everything returned is a copy: Task 6's |
| +// snapshot serializes this while probes keep running. |
| +func (p *proberSet) State() map[string][]UpstreamState { |
| + p.mu.Lock() |
| + hps := make(map[string]*hostProber, len(p.hosts)) |
| + for host, hp := range p.hosts { |
| + hps[host] = hp |
| + } |
| + p.mu.Unlock() |
| + |
| + out := make(map[string][]UpstreamState, len(hps)) |
| + for host, hp := range hps { |
| + hp.mu.Lock() |
| + states := make([]UpstreamState, 0, len(hp.order)) |
| + for _, addr := range hp.order { |
| + st := hp.state[addr] |
| + states = append(states, UpstreamState{ |
| + Addr: addr, |
| + Up: st.up, |
| + Since: st.since, |
| + Fails: st.fails, |
| + LastError: st.lastError, |
| + }) |
| + } |
| + hp.mu.Unlock() |
| + out[host] = states |
| + } |
| + return out |
| +} |
| + |
| +// Stop stops every prober and empties the set (tests and shutdown). Sync may |
| +// be called again afterwards; it simply starts fresh probers. |
| +func (p *proberSet) Stop() { |
| + p.mu.Lock() |
| + hps := make([]*hostProber, 0, len(p.hosts)) |
| + for host, hp := range p.hosts { |
| + hps = append(hps, hp) |
| + delete(p.hosts, host) |
| + } |
| + p.mu.Unlock() |
| + for _, hp := range hps { |
| + hp.stop() |
| + } |
| +} |
| + |
| +// run is one host's probe loop: probe every upstream, wait out the interval, |
| +// repeat. The first pass runs immediately so a dead upstream is caught within |
| +// fail_after × interval of the reload rather than one interval later. |
| +func (hp *hostProber) run() { |
| + defer close(hp.done) |
| + t := time.NewTicker(hp.interval) |
| + defer t.Stop() |
| + for { |
| + hp.tick() |
| + select { |
| + case <-hp.ctx.Done(): |
| + return |
| + case <-t.C: |
| + } |
| + } |
| +} |
| + |
| +// tick probes each upstream once, sequentially. Sequential is deliberate at |
| +// these counts (a route has a handful of upstreams, not hundreds): it bounds |
| +// the goroutine's own concurrency at one in-flight request and keeps the log |
| +// lines of a whole-fleet outage in addr order. The ctx checks are what keep |
| +// stop() prompt on a route with several dead addrs. |
| +// |
| +// hp.order is read without the lock on purpose: it is fixed by |
| +// newHostProber before this goroutine is started (an addr set change is a new |
| +// hostProber, never a mutation), so the `go hp.run()` in Sync is the |
| +// happens-before edge. Live/State still take the lock because they share it |
| +// with the state map. |
| +func (hp *hostProber) tick() { |
| + for _, addr := range hp.order { |
| + if hp.ctx.Err() != nil { |
| + return |
| + } |
| + err := hp.probe(addr) |
| + // A probe cancelled by our own stop says nothing about the upstream. |
| + // Recording it would let a retiring prober log a "down" transition |
| + // (and stamp a LastError) for a host that was merely reloaded. |
| + if hp.ctx.Err() != nil { |
| + return |
| + } |
| + hp.record(addr, err) |
| + } |
| +} |
| + |
| +// probe performs one health request. A nil error means the upstream answered |
| +// the configured status; anything else is the reason string that reaches both |
| +// the transition log and the dashboard's LastError. |
| +func (hp *hostProber) probe(addr string) error { |
| + req, err := http.NewRequestWithContext(hp.ctx, http.MethodGet, "http://"+addr+hp.path, nil) |
| + if err != nil { |
| + return fmt.Errorf("bad probe target: %v", err) |
| + } |
| + // Host, not the header map: net/http reads Request.Host for the wire |
| + // Host line, and an upstream that vhosts (or redirects on an unexpected |
| + // name) must see the name the operator configured. |
| + req.Host = hp.hostHdr |
| + resp, err := hp.client.Do(req) |
| + if err != nil { |
| + // Connect refused, timeout, reset, TLS-less port answering garbage: |
| + // from the edge's side these are one thing — no usable response |
| + // arrived — and the spec names that class "dial". |
| + return fmt.Errorf("dial: %v", err) |
| + } |
| + // Drain a bounded amount so the connection can be reused; a probe body is |
| + // expected to be tiny and is never inspected. |
| + io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) |
| + resp.Body.Close() |
| + if resp.StatusCode != hp.wantCode { |
| + return fmt.Errorf("status %d (want %d)", resp.StatusCode, hp.wantCode) |
| + } |
| + return nil |
| +} |
| + |
| +// record folds one probe result into an upstream's state and logs transitions |
| +// with their reason. Consecutive counts only: a single pass clears the failure |
| +// run and a single failure clears the recovery run, which is what makes |
| +// fail_after/rise_after mean "sustained", not "cumulative". |
| +func (hp *hostProber) record(addr string, probeErr error) { |
| + hp.mu.Lock() |
| + st := hp.state[addr] |
| + if st == nil { // addr vanished from this prober — impossible today |
| + hp.mu.Unlock() |
| + return |
| + } |
| + var transition, reason string |
| + switch { |
| + case probeErr == nil: |
| + st.fails = 0 |
| + st.lastError = "" |
| + if !st.up { |
| + st.passes++ |
| + if st.passes >= hp.riseAt { |
| + st.up = true |
| + st.passes = 0 |
| + st.since = time.Now() |
| + transition = "up" |
| + } |
| + } |
| + default: |
| + st.passes = 0 |
| + st.fails++ |
| + st.lastError = probeErr.Error() |
| + if st.up && st.fails >= hp.failAt { |
| + st.up = false |
| + st.since = time.Now() |
| + transition, reason = "down", st.lastError |
| + } |
| + } |
| + hp.mu.Unlock() |
| + |
| + switch transition { |
| + case "down": |
| + log.Printf("edge: probe %s %s down (%s)", hp.spec.Host, addr, reason) |
| + case "up": |
| + log.Printf("edge: probe %s %s up", hp.spec.Host, addr) |
| + } |
| +} |
amadan