| 1 | package edge |
| 2 | |
| 3 | // prober.go — active health probing for multi-upstream addr routes (HA design |
| 4 | // Part II, issue #152). One goroutine per probed host GETs each of its |
| 5 | // upstreams on a fixed cadence and maintains the LIVE SET the request path |
| 6 | // (multiProxy) picks from; the same table is the dashboard's per-upstream |
| 7 | // state. |
| 8 | // |
| 9 | // Two rules shape everything below: |
| 10 | // |
| 11 | // 1. **Optimistic start.** Every upstream of a newly synced host starts LIVE. |
| 12 | // A reload happens on any registry fingerprint move — an unrelated route |
| 13 | // changing must never take a healthy fleet off the air for the first |
| 14 | // fail_after × interval while probes catch up. Probing only ever REMOVES |
| 15 | // confidence; it never withholds it. |
| 16 | // |
| 17 | // 2. **Probes dial upstreams directly**, never the edge's own public host, so |
| 18 | // a probe cannot enter the activator and cannot wake a hibernating tenant |
| 19 | // — the wake hazard that would otherwise make active probing incompatible |
| 20 | // with hibernation. |
| 21 | // |
| 22 | // The second half of that guarantee is a MINT-SIDE door, not Reload's: |
| 23 | // provision.ValidateShape refuses `addrs` together with `hibernate: true` |
| 24 | // outright (internal/provision/provision.go, "addr with hibernate"), so a |
| 25 | // probed route is never activator-managed in the first place. Reload's |
| 26 | // `hib` gate is NOT that door — it reads `Kind == "instance" && Hibernate |
| 27 | // && !viaTunnel` and says nothing about addrs — so a hand-written registry |
| 28 | // row could still pair the two, and would then hibernate a process whose |
| 29 | // upstream the probe keeps marking down. Fixing that means widening the |
| 30 | // gate, not trusting this comment. |
| 31 | |
| 32 | import ( |
| 33 | "context" |
| 34 | "fmt" |
| 35 | "io" |
| 36 | "log" |
| 37 | "net/http" |
| 38 | "slices" |
| 39 | "sync" |
| 40 | "time" |
| 41 | |
| 42 | "github.com/carlosframework/platform/internal/provision" |
| 43 | "github.com/carlosframework/platform/internal/status" |
| 44 | ) |
| 45 | |
| 46 | // probeTimeout bounds a single probe request end to end (connect through |
| 47 | // body). Held at 2s per the spec: long enough that a loaded-but-alive |
| 48 | // upstream still answers, short enough that a black-holed one cannot stall a |
| 49 | // tick past the default 2s interval. |
| 50 | const probeTimeout = 2 * time.Second |
| 51 | |
| 52 | // Defensive fallbacks for a spec that reaches the prober with a non-positive |
| 53 | // cadence or count. Sync runs every incoming spec through |
| 54 | // provision.Health.WithDefaults, which covers the ZEROES; these clamps exist |
| 55 | // for what WithDefaults lets through — NEGATIVES — because mint is not the only |
| 56 | // writer of a route's health JSON (registry.SetUpstreams writes the column |
| 57 | // directly, and the route PATCH endpoint will too), and neither hand-edited |
| 58 | // JSON nor a future writer is obliged to be sane. |
| 59 | // |
| 60 | // The interval clamp is the load-bearing one: time.NewTicker PANICS on a |
| 61 | // non-positive duration, so an unclamped interval_ms of -1 (or 0, on any path |
| 62 | // that reached newHostProber without normalization) would take the whole edge |
| 63 | // process down — every route on the box, not just the misconfigured one. A |
| 64 | // non-positive fail_after must likewise not fail an upstream away on a single |
| 65 | // blip, nor a non-positive rise_after rejoin it on a single pass. |
| 66 | // |
| 67 | // Only NON-POSITIVE values are replaced. A deliberately fast interval is the |
| 68 | // operator's call (and the tests' — they probe at 20ms). |
| 69 | const ( |
| 70 | fallbackProbeInterval = 2000 * time.Millisecond |
| 71 | fallbackFailAfter = 3 |
| 72 | fallbackRiseAfter = 2 |
| 73 | ) |
| 74 | |
| 75 | // UpstreamState is the dashboard-facing view of one upstream: what the probe |
| 76 | // currently believes, when that belief last changed, and why. |
| 77 | // |
| 78 | // An ALIAS, not a declaration: the struct itself lives in internal/status |
| 79 | // (Task 6), because it is a wire type on status/boxes/<label>.json and its |
| 80 | // readers — the console, operators — must be able to decode it without |
| 81 | // importing the edge. See status.UpstreamState's own doc comment for the full |
| 82 | // argument. The alias keeps this package writing plain UpstreamState with no |
| 83 | // conversion where the prober fills it in. |
| 84 | type UpstreamState = status.UpstreamState |
| 85 | |
| 86 | // probeSpec is one route's probing contract, parsed from the registry's health |
| 87 | // JSON once per Reload. Health need not have been through WithDefaults: Sync |
| 88 | // normalizes it, so a caller may hand over the raw parse. |
| 89 | type probeSpec struct { |
| 90 | Host string // route host — the probe's Host header default |
| 91 | Addrs []string |
| 92 | Health provision.Health |
| 93 | } |
| 94 | |
| 95 | // equal reports whether two specs describe the same probing job. Sync uses it |
| 96 | // to decide restart-or-leave-alone, so it must cover every field the running |
| 97 | // goroutine captured: host (the default Host header), the addr list INCLUDING |
| 98 | // order (round-robin walks it), and the whole health object. Sync compares |
| 99 | // NORMALIZED specs, so Health{} and Health{}.WithDefaults() are equal here. |
| 100 | func (s probeSpec) equal(o probeSpec) bool { |
| 101 | return s.Host == o.Host && s.Health == o.Health && slices.Equal(s.Addrs, o.Addrs) |
| 102 | } |
| 103 | |
| 104 | // proberSet owns one prober goroutine per probed host plus the live-state |
| 105 | // table both the request path and the status snapshot read. |
| 106 | // |
| 107 | // Lifecycle mirrors Activator.RetireExcept: Reload computes the desired specs |
| 108 | // and calls Sync, which starts probers for new hosts, stops those whose route |
| 109 | // vanished, and replaces those whose spec changed. It is the only other thing |
| 110 | // in this package that owns per-route goroutines across a reload. |
| 111 | // |
| 112 | // A *proberSet must come from newProberSet; the zero value is not usable and a |
| 113 | // nil receiver panics rather than quietly reporting "no live upstreams" — a |
| 114 | // silent nil would turn a wiring mistake into fleet-wide 503s instead of a |
| 115 | // loud failure in the first test that hits it. |
| 116 | type proberSet struct { |
| 117 | // mu guards hosts only. The per-host state table lives on hostProber |
| 118 | // under its own lock, so a probe tick writing state never contends with |
| 119 | // Sync, and Sync never has to hold this lock while waiting for a |
| 120 | // goroutine to exit (which would stall Live on the request hot path). |
| 121 | mu sync.Mutex |
| 122 | hosts map[string]*hostProber |
| 123 | client *http.Client |
| 124 | } |
| 125 | |
| 126 | func newProberSet() *proberSet { |
| 127 | return &proberSet{ |
| 128 | hosts: make(map[string]*hostProber), |
| 129 | client: &http.Client{ |
| 130 | Timeout: probeTimeout, |
| 131 | // A redirect is not health: an upstream answering 302 has not |
| 132 | // served the configured status, and following the hop would |
| 133 | // probe something the operator never named. |
| 134 | CheckRedirect: func(*http.Request, []*http.Request) error { |
| 135 | return http.ErrUseLastResponse |
| 136 | }, |
| 137 | Transport: &http.Transport{ |
| 138 | // Probes are the same tiny request to the same handful of |
| 139 | // addrs forever, so idle connections are worth keeping — |
| 140 | // bounded tightly, since every probed upstream holds one. |
| 141 | MaxIdleConnsPerHost: 1, |
| 142 | IdleConnTimeout: 90 * time.Second, |
| 143 | ResponseHeaderTimeout: probeTimeout, |
| 144 | }, |
| 145 | }, |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | // hostProber is one host's probing job: its spec, the clamped cadence derived |
| 150 | // from it, its goroutine's stop handle, and the per-addr state that goroutine |
| 151 | // maintains. |
| 152 | // |
| 153 | // The state table is OWNED BY THIS STRUCT, not by the set — that is what makes |
| 154 | // Sync's replace safe. A replaced prober's goroutine can still be mid-probe |
| 155 | // when the new one is installed, and whatever it writes lands in its own |
| 156 | // retired table, which nothing reads any more. (Sync additionally waits for |
| 157 | // the old goroutine to observe its stop before starting the new one, so even |
| 158 | // the log lines stay ordered.) |
| 159 | type hostProber struct { |
| 160 | spec probeSpec |
| 161 | client *http.Client |
| 162 | path string |
| 163 | wantCode int |
| 164 | hostHdr string |
| 165 | interval time.Duration |
| 166 | failAt int |
| 167 | riseAt int |
| 168 | |
| 169 | cancel context.CancelFunc |
| 170 | ctx context.Context |
| 171 | done chan struct{} |
| 172 | |
| 173 | mu sync.Mutex |
| 174 | order []string // addrs in spec order — Live/State iterate this, not the map |
| 175 | state map[string]*upstreamState |
| 176 | } |
| 177 | |
| 178 | // upstreamState is the mutable half of UpstreamState plus the pass counter, |
| 179 | // which is bookkeeping the dashboard has no use for. |
| 180 | type upstreamState struct { |
| 181 | up bool |
| 182 | since time.Time |
| 183 | fails int |
| 184 | passes int |
| 185 | lastError string |
| 186 | } |
| 187 | |
| 188 | func newHostProber(spec probeSpec, client *http.Client) *hostProber { |
| 189 | h := spec.Health |
| 190 | hp := &hostProber{ |
| 191 | spec: spec, |
| 192 | client: client, |
| 193 | path: h.Path, |
| 194 | wantCode: h.Status, |
| 195 | hostHdr: h.HostHeader, |
| 196 | interval: time.Duration(h.IntervalMS) * time.Millisecond, |
| 197 | failAt: h.FailAfter, |
| 198 | riseAt: h.RiseAfter, |
| 199 | done: make(chan struct{}), |
| 200 | state: make(map[string]*upstreamState, len(spec.Addrs)), |
| 201 | } |
| 202 | if hp.path == "" { |
| 203 | hp.path = "/" |
| 204 | } |
| 205 | if hp.path[0] != '/' { |
| 206 | hp.path = "/" + hp.path |
| 207 | } |
| 208 | if hp.wantCode == 0 { |
| 209 | hp.wantCode = http.StatusOK |
| 210 | } |
| 211 | if hp.hostHdr == "" { |
| 212 | // WithDefaults deliberately leaves HostHeader empty so this |
| 213 | // substitution can see "unset" rather than a value it invented. |
| 214 | hp.hostHdr = spec.Host |
| 215 | } |
| 216 | if h.IntervalMS <= 0 { |
| 217 | hp.interval = fallbackProbeInterval |
| 218 | } |
| 219 | if hp.failAt <= 0 { |
| 220 | hp.failAt = fallbackFailAfter |
| 221 | } |
| 222 | if hp.riseAt <= 0 { |
| 223 | hp.riseAt = fallbackRiseAfter |
| 224 | } |
| 225 | now := time.Now() |
| 226 | for _, addr := range spec.Addrs { |
| 227 | if addr == "" || hp.state[addr] != nil { |
| 228 | continue // validation refuses both; be inert if one slips through |
| 229 | } |
| 230 | hp.order = append(hp.order, addr) |
| 231 | // Rule 1: everything starts live. |
| 232 | hp.state[addr] = &upstreamState{up: true, since: now} |
| 233 | } |
| 234 | hp.ctx, hp.cancel = context.WithCancel(context.Background()) |
| 235 | return hp |
| 236 | } |
| 237 | |
| 238 | // stop cancels the goroutine's context — aborting any in-flight probe at once, |
| 239 | // so this does not wait out the 2s client timeout — and blocks until it has |
| 240 | // exited. |
| 241 | func (hp *hostProber) stop() { |
| 242 | hp.cancel() |
| 243 | <-hp.done |
| 244 | } |
| 245 | |
| 246 | // Sync starts, stops and replaces probers so the running set matches want. |
| 247 | // Called from Reload after the upstream map swap, beside RetireExcept. |
| 248 | // |
| 249 | // Specs with no addrs are not probed (nothing to probe); such a host holds no |
| 250 | // entry, and Live/State report it as absent. |
| 251 | // |
| 252 | // Concurrent callers are safe (the set lock serializes the diff), but be aware |
| 253 | // that a call can BLOCK on stop()ing a retired prober — including one a peer |
| 254 | // call has installed but not yet started, since the goroutine launch happens |
| 255 | // after the lock is released. Reload is the single caller today, so this is a |
| 256 | // note for whoever adds the second one, not a live hazard. |
| 257 | func (p *proberSet) Sync(want map[string]probeSpec) { |
| 258 | // Normalize BEFORE both the compare and the construction. A caller that |
| 259 | // hands over a raw Health{} on one Reload and a WithDefaults()ed one on the |
| 260 | // next is describing the same probing job; comparing them un-normalized |
| 261 | // would call that a change and restart every prober on the box — resetting |
| 262 | // every fail/rise run — on each Reload a mixed-path writer triggers. |
| 263 | norm := make(map[string]probeSpec, len(want)) |
| 264 | for host, w := range want { |
| 265 | w.Health = w.Health.WithDefaults() |
| 266 | norm[host] = w |
| 267 | } |
| 268 | want = norm |
| 269 | |
| 270 | var retired, started []*hostProber |
| 271 | |
| 272 | p.mu.Lock() |
| 273 | for host, hp := range p.hosts { |
| 274 | w, ok := want[host] |
| 275 | if ok && len(w.Addrs) > 0 && hp.spec.equal(w) { |
| 276 | continue // unchanged: leave the goroutine and its counters alone |
| 277 | } |
| 278 | retired = append(retired, hp) |
| 279 | delete(p.hosts, host) |
| 280 | } |
| 281 | for host, w := range want { |
| 282 | if len(w.Addrs) == 0 { |
| 283 | continue |
| 284 | } |
| 285 | if _, ok := p.hosts[host]; ok { |
| 286 | continue |
| 287 | } |
| 288 | // Built (and therefore all-live) INSIDE the same critical section |
| 289 | // that removed the outgoing prober, so a concurrent Live never |
| 290 | // observes a replaced host as having no upstreams — a gap there |
| 291 | // would be a 503 for a route that is perfectly healthy. |
| 292 | hp := newHostProber(w, p.client) |
| 293 | p.hosts[host] = hp |
| 294 | started = append(started, hp) |
| 295 | } |
| 296 | p.mu.Unlock() |
| 297 | |
| 298 | // Outside the lock, and before the replacements run: retiring waits for |
| 299 | // each old goroutine to exit, and blocking Live (request hot path) for |
| 300 | // that wait would be exactly the stall this whole feature exists to |
| 301 | // avoid. |
| 302 | for _, hp := range retired { |
| 303 | hp.stop() |
| 304 | } |
| 305 | for _, hp := range started { |
| 306 | go hp.run() |
| 307 | } |
| 308 | } |
| 309 | |
| 310 | // Live reports the currently-live upstreams for host, in the route's addr |
| 311 | // order, as a fresh slice the caller may keep. Unknown host (or one with no |
| 312 | // live upstream) → empty: the request path answers 503 + Retry-After rather |
| 313 | // than dialing something the probe says is dead. |
| 314 | func (p *proberSet) Live(host string) []string { |
| 315 | p.mu.Lock() |
| 316 | hp := p.hosts[host] |
| 317 | p.mu.Unlock() |
| 318 | if hp == nil { |
| 319 | return nil |
| 320 | } |
| 321 | hp.mu.Lock() |
| 322 | defer hp.mu.Unlock() |
| 323 | out := make([]string, 0, len(hp.order)) |
| 324 | for _, addr := range hp.order { |
| 325 | if hp.state[addr].up { |
| 326 | out = append(out, addr) |
| 327 | } |
| 328 | } |
| 329 | return out |
| 330 | } |
| 331 | |
| 332 | // Eject drops addr from host's live set immediately — the in-request |
| 333 | // dial-failure fast path, so a box that died between probe ticks costs one |
| 334 | // retry rather than up to fail_after × interval of 502s. Fails is set to |
| 335 | // failAfter so the upstream is exactly as down as a probe-failed one; probe |
| 336 | // passes rejoin it after rise_after, the same as any other recovery. |
| 337 | // |
| 338 | // A host or addr this set does not know is a no-op: the request path can |
| 339 | // legitimately race a Reload that just dropped either. |
| 340 | func (p *proberSet) Eject(host, addr string) { |
| 341 | p.mu.Lock() |
| 342 | hp := p.hosts[host] |
| 343 | p.mu.Unlock() |
| 344 | if hp == nil { |
| 345 | return |
| 346 | } |
| 347 | hp.mu.Lock() |
| 348 | st := hp.state[addr] |
| 349 | if st == nil { |
| 350 | hp.mu.Unlock() |
| 351 | return |
| 352 | } |
| 353 | const reason = "dial failed on a live request" |
| 354 | was := st.up |
| 355 | st.up = false |
| 356 | st.passes = 0 |
| 357 | st.fails = hp.failAt |
| 358 | st.lastError = reason |
| 359 | if was { |
| 360 | st.since = time.Now() |
| 361 | } |
| 362 | hp.mu.Unlock() |
| 363 | if was { |
| 364 | log.Printf("edge: probe %s %s down (%s)", host, addr, reason) |
| 365 | } |
| 366 | } |
| 367 | |
| 368 | // State returns the probe's view of every probed host, keyed by host, each |
| 369 | // upstream in the route's addr order. Everything returned is a copy: Task 6's |
| 370 | // snapshot serializes this while probes keep running. |
| 371 | func (p *proberSet) State() map[string][]UpstreamState { |
| 372 | p.mu.Lock() |
| 373 | hps := make(map[string]*hostProber, len(p.hosts)) |
| 374 | for host, hp := range p.hosts { |
| 375 | hps[host] = hp |
| 376 | } |
| 377 | p.mu.Unlock() |
| 378 | |
| 379 | out := make(map[string][]UpstreamState, len(hps)) |
| 380 | for host, hp := range hps { |
| 381 | hp.mu.Lock() |
| 382 | states := make([]UpstreamState, 0, len(hp.order)) |
| 383 | for _, addr := range hp.order { |
| 384 | st := hp.state[addr] |
| 385 | states = append(states, UpstreamState{ |
| 386 | Addr: addr, |
| 387 | Up: st.up, |
| 388 | Since: st.since, |
| 389 | Fails: st.fails, |
| 390 | LastError: st.lastError, |
| 391 | }) |
| 392 | } |
| 393 | hp.mu.Unlock() |
| 394 | out[host] = states |
| 395 | } |
| 396 | return out |
| 397 | } |
| 398 | |
| 399 | // Stop stops every prober and empties the set (tests and shutdown). Sync may |
| 400 | // be called again afterwards; it simply starts fresh probers. |
| 401 | func (p *proberSet) Stop() { |
| 402 | p.mu.Lock() |
| 403 | hps := make([]*hostProber, 0, len(p.hosts)) |
| 404 | for host, hp := range p.hosts { |
| 405 | hps = append(hps, hp) |
| 406 | delete(p.hosts, host) |
| 407 | } |
| 408 | p.mu.Unlock() |
| 409 | for _, hp := range hps { |
| 410 | hp.stop() |
| 411 | } |
| 412 | // Every probed upstream held an idle connection (MaxIdleConnsPerHost 1); |
| 413 | // with no prober left to reuse them they are just sockets on a box that was |
| 414 | // told to stop. A later Sync gets fresh ones. |
| 415 | p.client.CloseIdleConnections() |
| 416 | } |
| 417 | |
| 418 | // run is one host's probe loop: probe every upstream, wait out the interval, |
| 419 | // repeat. The first pass runs immediately so a dead upstream is caught within |
| 420 | // fail_after × interval of the reload rather than one interval later. |
| 421 | func (hp *hostProber) run() { |
| 422 | defer close(hp.done) |
| 423 | t := time.NewTicker(hp.interval) |
| 424 | defer t.Stop() |
| 425 | for { |
| 426 | hp.tick() |
| 427 | select { |
| 428 | case <-hp.ctx.Done(): |
| 429 | return |
| 430 | case <-t.C: |
| 431 | } |
| 432 | } |
| 433 | } |
| 434 | |
| 435 | // tick probes each upstream once, sequentially. Sequential is deliberate at |
| 436 | // these counts (a route has a handful of upstreams, not hundreds): it bounds |
| 437 | // the goroutine's own concurrency at one in-flight request and keeps the log |
| 438 | // lines of a whole-fleet outage in addr order. The ctx checks are what keep |
| 439 | // stop() prompt on a route with several dead addrs. |
| 440 | // |
| 441 | // Under a TOTAL outage a tick can outlast the interval: N black-holed addrs × |
| 442 | // the 2s probe timeout against, say, a 2s cadence. time.Ticker has a one-slot |
| 443 | // buffer, so the missed ticks collapse into one and the loop simply runs |
| 444 | // back-to-back rather than queueing a backlog. That is the intended |
| 445 | // degradation — the probe rate self-limits to "as fast as the timeouts allow" |
| 446 | // and returns to cadence the moment upstreams answer again. |
| 447 | // |
| 448 | // hp.order is read without the lock on purpose: it is fixed by |
| 449 | // newHostProber before this goroutine is started (an addr set change is a new |
| 450 | // hostProber, never a mutation), so the `go hp.run()` in Sync is the |
| 451 | // happens-before edge. Live/State still take the lock because they share it |
| 452 | // with the state map. |
| 453 | func (hp *hostProber) tick() { |
| 454 | for _, addr := range hp.order { |
| 455 | if hp.ctx.Err() != nil { |
| 456 | return |
| 457 | } |
| 458 | err := hp.probe(addr) |
| 459 | // A probe cancelled by our own stop says nothing about the upstream. |
| 460 | // Recording it would let a retiring prober log a "down" transition |
| 461 | // (and stamp a LastError) for a host that was merely reloaded. |
| 462 | if hp.ctx.Err() != nil { |
| 463 | return |
| 464 | } |
| 465 | hp.record(addr, err) |
| 466 | } |
| 467 | } |
| 468 | |
| 469 | // probe performs one health request. A nil error means the upstream answered |
| 470 | // the configured status; anything else is the reason string that reaches both |
| 471 | // the transition log and the dashboard's LastError. |
| 472 | func (hp *hostProber) probe(addr string) error { |
| 473 | req, err := http.NewRequestWithContext(hp.ctx, http.MethodGet, "http://"+addr+hp.path, nil) |
| 474 | if err != nil { |
| 475 | return fmt.Errorf("bad probe target: %v", err) |
| 476 | } |
| 477 | // Host, not the header map: net/http reads Request.Host for the wire |
| 478 | // Host line, and an upstream that vhosts (or redirects on an unexpected |
| 479 | // name) must see the name the operator configured. |
| 480 | req.Host = hp.hostHdr |
| 481 | resp, err := hp.client.Do(req) |
| 482 | if err != nil { |
| 483 | // Connect refused, timeout, reset, TLS-less port answering garbage: |
| 484 | // from the edge's side these are one thing — no usable response |
| 485 | // arrived — and the spec names that class "dial". |
| 486 | return fmt.Errorf("dial: %v", err) |
| 487 | } |
| 488 | // Drain a bounded amount so the connection can be reused; a probe body is |
| 489 | // expected to be tiny and is never inspected. |
| 490 | io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) |
| 491 | resp.Body.Close() |
| 492 | if resp.StatusCode != hp.wantCode { |
| 493 | return fmt.Errorf("status %d (want %d)", resp.StatusCode, hp.wantCode) |
| 494 | } |
| 495 | return nil |
| 496 | } |
| 497 | |
| 498 | // record folds one probe result into an upstream's state and logs transitions |
| 499 | // with their reason. Consecutive counts only: a single pass clears the failure |
| 500 | // run and a single failure clears the recovery run, which is what makes |
| 501 | // fail_after/rise_after mean "sustained", not "cumulative". |
| 502 | // |
| 503 | // Fails and LastError are the DOWN upstream's explanation, so they survive the |
| 504 | // passing probes of a recovery and are cleared only on the up transition. Zero |
| 505 | // them on the first pass instead and the dashboard shows the mid-recovery |
| 506 | // upstream as `Up:false, Fails:0, LastError:""` — down for no stated reason, |
| 507 | // which is the one thing an operator staring at a failover must not see. |
| 508 | func (hp *hostProber) record(addr string, probeErr error) { |
| 509 | hp.mu.Lock() |
| 510 | st := hp.state[addr] |
| 511 | if st == nil { // addr vanished from this prober — impossible today |
| 512 | hp.mu.Unlock() |
| 513 | return |
| 514 | } |
| 515 | var transition, reason string |
| 516 | switch { |
| 517 | case probeErr == nil: |
| 518 | if st.up { |
| 519 | // Healthy and staying healthy: the failure run ends here. |
| 520 | st.fails = 0 |
| 521 | st.lastError = "" |
| 522 | break |
| 523 | } |
| 524 | st.passes++ |
| 525 | if st.passes >= hp.riseAt { |
| 526 | st.up = true |
| 527 | st.passes = 0 |
| 528 | st.fails = 0 |
| 529 | st.lastError = "" |
| 530 | st.since = time.Now() |
| 531 | transition = "up" |
| 532 | } |
| 533 | default: |
| 534 | st.passes = 0 |
| 535 | st.fails++ |
| 536 | st.lastError = probeErr.Error() |
| 537 | if st.up && st.fails >= hp.failAt { |
| 538 | st.up = false |
| 539 | st.since = time.Now() |
| 540 | transition, reason = "down", st.lastError |
| 541 | } |
| 542 | } |
| 543 | hp.mu.Unlock() |
| 544 | |
| 545 | switch transition { |
| 546 | case "down": |
| 547 | log.Printf("edge: probe %s %s down (%s)", hp.spec.Host, addr, reason) |
| 548 | case "up": |
| 549 | log.Printf("edge: probe %s %s up", hp.spec.Host, addr) |
| 550 | } |
| 551 | } |
| 552 | |
amadan