| 1 | package main |
| 2 | |
| 3 | // The activator mesh (issue #550 Part 4 v2, Phase 3): multiple router boxes |
| 4 | // cooperate so a hibernating instance wakes on exactly one box, a request |
| 5 | // landing on the wrong box is reverse-proxied to the owner, and a dead box's |
| 6 | // instances wake elsewhere. Four mechanisms, deliberately this simple: |
| 7 | // |
| 8 | // 1. Membership — boxes learn who's alive from S3 heartbeat objects |
| 9 | // (internal/activator/heartbeat.go), not gossip: the bucket is already the |
| 10 | // shared substrate, so no new ports or protocols between boxes (which may |
| 11 | // not even share a network). Each box PUTs mesh/members/<box> every |
| 12 | // meshHeartbeatInterval and LISTs the prefix; a record older than |
| 13 | // meshStaleAfter is a dead box. A box's identity is LCHAT_MESH_SELF — a |
| 14 | // base URL its peers can reach it at; the env's presence turns the mesh |
| 15 | // on, its absence keeps today's single-box behavior byte-identical. |
| 16 | // 2. Rendezvous (HRW) hashing — owner(host) is a pure function of the live |
| 17 | // member set (hrwOwner): no routing table, no directory service, and a |
| 18 | // membership change moves only ~1/N of hosts (only onto a joiner, only |
| 19 | // off a leaver). |
| 20 | // 3. The lease stays the truth; hashing is only a hint. The owner claims the |
| 21 | // per-host S3 lease exactly as on one box; a box that wrongly believes |
| 22 | // it's the owner loses the CAS race loudly (503 + Retry-After) and the |
| 23 | // client's retry lands right. Transient membership disagreement therefore |
| 24 | // cannot split-brain — single-writer was never the mesh's job. |
| 25 | // 4. Proxy-to-owner — a request for a sleeping host whose HRW owner is |
| 26 | // another live box is reverse-proxied to that box's peer URL, instance |
| 27 | // hostname preserved (the peer routes by Host like any front door). A |
| 28 | // dead/stale owner drops out of the member set, so the hash reassigns and |
| 29 | // the request falls through to a local wake — failover IS the routine |
| 30 | // wake path. |
| 31 | // 5. Domain scoping (LCHAT_MESH_DOMAINS) — HRW placement only ever applies to |
| 32 | // hosts under a configured mesh domain; everything else always resolves |
| 33 | // locally, on whichever box is asked. This isn't an optimization: a |
| 34 | // registry row only ever exists on the boxes it was explicitly replicated |
| 35 | // to (see registerInstance in control.go), so a host outside the |
| 36 | // configured domains was never given a row anywhere but its original box, |
| 37 | // and letting the hash "place" it elsewhere would just 404 there. It also |
| 38 | // keeps placement bounded to hosts every mesh member can actually get a |
| 39 | // matching TLS cert for (typically the same domains as |
| 40 | // LCHAT_WILDCARD_DOMAINS's ACME DNS-01 cert, which — unlike per-host |
| 41 | // HTTP-01 — any box can obtain independently of which box DNS currently |
| 42 | // points at). |
| 43 | // 6. Region placement (LCHAT_MESH_REGION + a space's home_region) — the |
| 44 | // multi-edge foundation (docs/multi-edge.md, Phase 1). Within the in-scope |
| 45 | // set, HRW picks the owner from only the live boxes whose region matches |
| 46 | // the space's home_region (candidatesFor), so a space lives on a box in its |
| 47 | // region and relocating it is just a home_region flip (the hash re-homes it |
| 48 | // onto the destination box, the lease still guaranteeing one writer). A |
| 49 | // space with no home_region, OR whose region currently has no live box, |
| 50 | // falls back to hashing across the WHOLE live set — byte-identical to |
| 51 | // pre-region placement, which is what keeps today's single-region fleet |
| 52 | // (and every self-hoster) unchanged. Boxes learn each other's region from |
| 53 | // the heartbeat body (heartbeat.go's Regions); the home_region itself is |
| 54 | // replicated into every box's registry alongside the route (meshreplicate.go). |
| 55 | // |
| 56 | // Drain (activatorSvc.shutdown) deletes the heartbeat FIRST, then checkpoints |
| 57 | // and releases as today — peers take over immediately, and deploys, box |
| 58 | // failure, and scale-down are the same code path. |
| 59 | |
| 60 | import ( |
| 61 | "context" |
| 62 | "errors" |
| 63 | "fmt" |
| 64 | "hash/fnv" |
| 65 | "io" |
| 66 | "log" |
| 67 | "net" |
| 68 | "net/http" |
| 69 | "net/http/httputil" |
| 70 | "net/url" |
| 71 | "os" |
| 72 | "strings" |
| 73 | "sync" |
| 74 | "time" |
| 75 | |
| 76 | "lchat/internal/activator" |
| 77 | ) |
| 78 | |
| 79 | // meshForwardedHeader marks a request one box already proxied to another: |
| 80 | // "serve where you stand, never re-proxy". One hop max — if two boxes |
| 81 | // transiently disagree about ownership, the worst case is a wrong-box wake |
| 82 | // the lease arbitrates, never a proxy loop. An outsider spoofing it merely |
| 83 | // pins its own request to the box it hit, which is exactly what every request |
| 84 | // could do before the mesh existed; the lease still guarantees one writer. |
| 85 | // Stripped before any backend sees it (unixProxy), like the sidecar marker. |
| 86 | const meshForwardedHeader = "X-Lchat-Mesh-Forwarded" |
| 87 | |
| 88 | const ( |
| 89 | // meshHeartbeatInterval paces the PUT + LIST beat. Membership changes are |
| 90 | // rare events; this is liveness, not coordination — and each beat is a PUT |
| 91 | // plus a membership LIST, and now a region refresh (heartbeat.go's Regions: |
| 92 | // a second LIST plus one small GET per live peer — a handful of requests at |
| 93 | // fleet scale), so it stays cheap for nothing membership-shaped in return. |
| 94 | // Slower still would work, but |
| 95 | // staleness (3 beats) rides on this, and 3m dead-box detection already |
| 96 | // sits well inside the lease expiry that actually gates takeover. |
| 97 | meshHeartbeatInterval = 60 * time.Second |
| 98 | // meshStaleAfter is how old a heartbeat may be before its box is presumed |
| 99 | // dead — three missed beats. It's judged against the reader's clock but |
| 100 | // the timestamp is S3's LastModified (the bucket's clock), so only |
| 101 | // reader-vs-S3 skew matters, tiny against 3m. Together with the lease |
| 102 | // TTL (activator.LeaseTTL, 15m) this bounds failover: a hard-killed box's |
| 103 | // hosts wake elsewhere once its heartbeat is stale AND its leases expire — |
| 104 | // lease-dominated, so ~15m. |
| 105 | meshStaleAfter = 3 * meshHeartbeatInterval |
| 106 | ) |
| 107 | |
| 108 | // meshStore is the S3 heartbeat surface (activator.HeartbeatStore in |
| 109 | // production), an interface so the mesh's unit tests share an in-memory |
| 110 | // "bucket" between boxes. |
| 111 | type meshStore interface { |
| 112 | Put(ctx context.Context, peer, region string) error |
| 113 | List(ctx context.Context) ([]activator.Heartbeat, error) |
| 114 | // Regions returns each peer's advertised placement region (peer URL -> |
| 115 | // region); a peer absent from the map advertises none (agnostic). Read off |
| 116 | // the hot path, on the beat — see heartbeat.go's Regions. |
| 117 | Regions(ctx context.Context) (map[string]string, error) |
| 118 | Delete(ctx context.Context, peer string) error |
| 119 | } |
| 120 | |
| 121 | type meshSvc struct { |
| 122 | self string // this box's peer base URL — its mesh identity |
| 123 | store meshStore |
| 124 | domains []string // LCHAT_MESH_DOMAINS, lowercased — see inScope. Empty = unrestricted (test convenience only; meshFromEnv always supplies a non-empty list) |
| 125 | token string // LCHAT_MESH_TOKEN — shared secret authenticating box-to-box route replication (meshreplicate.go); empty = unrestricted (test convenience only) |
| 126 | region string // LCHAT_MESH_REGION — this box's placement region (e.g. "ie"); "" = region-agnostic (today's behaviour). Advertised in the heartbeat body; peers read it via Regions. See candidatesFor. |
| 127 | reg *Registry // local registry handle, for reading a host's replicated home_region (candidatesFor). nil = no lookup (direct-newMesh unit tests) = every space region-agnostic. |
| 128 | interval time.Duration |
| 129 | staleAfter time.Duration |
| 130 | now func() time.Time // a seam for the unit tests' clock |
| 131 | |
| 132 | mu sync.Mutex |
| 133 | members []activator.Heartbeat // last LIST; staleness applied at read time |
| 134 | regions map[string]string // last Regions view: peer URL -> advertised region (self excluded — known locally via m.region) |
| 135 | proxies map[string]*httputil.ReverseProxy |
| 136 | started bool |
| 137 | |
| 138 | stopCh chan struct{} |
| 139 | done chan struct{} |
| 140 | } |
| 141 | |
| 142 | func newMesh(self string, store meshStore, domains []string) *meshSvc { |
| 143 | return &meshSvc{ |
| 144 | self: self, |
| 145 | store: store, |
| 146 | domains: domains, |
| 147 | interval: meshHeartbeatInterval, |
| 148 | staleAfter: meshStaleAfter, |
| 149 | now: time.Now, |
| 150 | regions: make(map[string]string), |
| 151 | proxies: make(map[string]*httputil.ReverseProxy), |
| 152 | stopCh: make(chan struct{}), |
| 153 | done: make(chan struct{}), |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | // meshFromEnv reads the box's mesh identity. LCHAT_MESH_SELF is a base URL |
| 158 | // peer boxes can reach this box at, scheme included — http on a private |
| 159 | // network, or https where the box presents a verifiable cert (see |
| 160 | // newPeerProxy). Presence turns the mesh on; absent = single box, zero mesh |
| 161 | // codepaths. The mesh rides on the hibernation bucket, so it requires |
| 162 | // LCHAT_HIBERNATE_BUCKET. It also requires LCHAT_MESH_DOMAINS (same |
| 163 | // comma-list shape as LCHAT_WILDCARD_DOMAINS, parsed by the same |
| 164 | // splitDomains): failing loudly here beats silently letting HRW "place" a |
| 165 | // host onto a box that has no registry row for it (a 404, not a graceful |
| 166 | // wake) or that could never get it a matching cert — see inScope. And |
| 167 | // LCHAT_MESH_TOKEN, the shared secret every mesh member authenticates |
| 168 | // box-to-box route replication with (meshreplicate.go) — without it any box |
| 169 | // could hand any other box arbitrary routes. |
| 170 | // |
| 171 | // LCHAT_MESH_REGION is optional: this box's placement region (e.g. "ie"). Set, |
| 172 | // it opts the box into region-scoped placement (candidatesFor); unset, the box |
| 173 | // is region-agnostic and everything behaves exactly as before regions existed. |
| 174 | func meshFromEnv(hc *hibernateConfig) (*meshSvc, error) { |
| 175 | self := strings.TrimSpace(os.Getenv("LCHAT_MESH_SELF")) |
| 176 | if self == "" { |
| 177 | return nil, nil |
| 178 | } |
| 179 | if hc == nil { |
| 180 | return nil, errors.New("LCHAT_MESH_SELF set but LCHAT_HIBERNATE_BUCKET missing — the mesh rides on the hibernation bucket and only places hibernating instances") |
| 181 | } |
| 182 | domains := splitDomains(os.Getenv("LCHAT_MESH_DOMAINS")) |
| 183 | if len(domains) == 0 { |
| 184 | return nil, errors.New("LCHAT_MESH_SELF set but LCHAT_MESH_DOMAINS missing — mesh placement needs an explicit domain allowlist so HRW never assigns a host to a box that has no registry row for it or can't obtain it a matching cert; set it to the domains every mesh member can serve (typically the same value as LCHAT_WILDCARD_DOMAINS)") |
| 185 | } |
| 186 | token := strings.TrimSpace(os.Getenv("LCHAT_MESH_TOKEN")) |
| 187 | if token == "" { |
| 188 | return nil, errors.New("LCHAT_MESH_SELF set but LCHAT_MESH_TOKEN missing — route replication between mesh members needs a shared secret; generate one and set it identically on every mesh box") |
| 189 | } |
| 190 | self = strings.TrimRight(self, "/") |
| 191 | u, err := url.Parse(self) |
| 192 | if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") || u.Path != "" { |
| 193 | return nil, fmt.Errorf("LCHAT_MESH_SELF %q: want a base URL like https://box-a.internal or http://10.0.0.5:8080", self) |
| 194 | } |
| 195 | store, err := activator.NewHeartbeatStore(context.Background(), hc.Bucket, hc.meshPrefix(), hc.Region) |
| 196 | if err != nil { |
| 197 | return nil, fmt.Errorf("mesh heartbeat store: %w", err) |
| 198 | } |
| 199 | m := newMesh(self, store, domains) |
| 200 | m.token = token |
| 201 | m.region = strings.TrimSpace(os.Getenv("LCHAT_MESH_REGION")) |
| 202 | return m, nil |
| 203 | } |
| 204 | |
| 205 | // start registers this box (one synchronous beat, so the first sweep and the |
| 206 | // first request already see a membership view) and begins the heartbeat loop. |
| 207 | func (m *meshSvc) start() { |
| 208 | m.beat() |
| 209 | m.mu.Lock() |
| 210 | m.started = true |
| 211 | m.mu.Unlock() |
| 212 | go func() { |
| 213 | defer close(m.done) |
| 214 | t := time.NewTicker(m.interval) |
| 215 | defer t.Stop() |
| 216 | for { |
| 217 | select { |
| 218 | case <-m.stopCh: |
| 219 | return |
| 220 | case <-t.C: |
| 221 | m.beat() |
| 222 | } |
| 223 | } |
| 224 | }() |
| 225 | } |
| 226 | |
| 227 | // beat PUTs our heartbeat (with our region) and refreshes the member view, plus |
| 228 | // the peer-region view WHEN this box has opted into region placement. Every |
| 229 | // failure is soft by design: a missed PUT makes peers see us stale (they take |
| 230 | // over; the lease arbitrates), a failed LIST keeps the last known membership (a |
| 231 | // stale view misroutes at worst — same arbitration), and a failed region |
| 232 | // refresh keeps the last regions (at worst a space scopes to the whole fleet |
| 233 | // for one beat, never wrong — candidatesFor's fallback IS all-boxes). |
| 234 | // |
| 235 | // Region placement is opt-in PER MESH: a box that declares no region |
| 236 | // (LCHAT_MESH_REGION unset — prod today) skips the peer-region fetch entirely, |
| 237 | // so it adds ZERO S3 requests over pre-region behaviour — an inert feature |
| 238 | // costs nothing. Such a box leaves m.regions empty and candidatesFor always |
| 239 | // falls back to all-boxes HRW. A mixed fleet (some boxes with regions, some |
| 240 | // without) is safe too: an agnostic box just can't region-scope, so it hashes |
| 241 | // across everyone, and any resulting placement disagreement is absorbed by the |
| 242 | // S3 single-writer lease exactly as ordinary membership disagreement is — a |
| 243 | // wrong-box wake, never corruption. |
| 244 | func (m *meshSvc) beat() { |
| 245 | timeout := m.interval |
| 246 | if timeout < 5*time.Second { |
| 247 | timeout = 5 * time.Second |
| 248 | } |
| 249 | ctx, cancel := context.WithTimeout(context.Background(), timeout) |
| 250 | defer cancel() |
| 251 | // Put stays unconditional — the heartbeat is membership (its key + the write |
| 252 | // timestamp), region-or-not; an unset region just writes an empty-region body. |
| 253 | if err := m.store.Put(ctx, m.self, m.region); err != nil { |
| 254 | log.Printf("mesh: heartbeat put failed: %v — peers will see this box stale after %s", err, m.staleAfter) |
| 255 | } |
| 256 | members, err := m.store.List(ctx) |
| 257 | if err != nil { |
| 258 | log.Printf("mesh: membership list failed: %v — keeping the last known view", err) |
| 259 | return |
| 260 | } |
| 261 | m.mu.Lock() |
| 262 | m.members = members |
| 263 | m.mu.Unlock() |
| 264 | if m.region == "" { |
| 265 | return // not region-scoped — no peer regions to fetch, zero extra S3 |
| 266 | } |
| 267 | // Region-scoped: refresh peers' regions. A box's region is static, so this |
| 268 | // per-beat re-fetch could later narrow to fetch-on-new-peer only (a |
| 269 | // planet-scale optimisation — out of scope for this slice, where the fleet |
| 270 | // is a handful of boxes and a per-beat GET each is noise). |
| 271 | regions, err := m.store.Regions(ctx) |
| 272 | if err != nil { |
| 273 | log.Printf("mesh: region refresh failed: %v — keeping the last known regions", err) |
| 274 | return |
| 275 | } |
| 276 | m.mu.Lock() |
| 277 | m.regions = regions |
| 278 | m.mu.Unlock() |
| 279 | } |
| 280 | |
| 281 | // halt stops the heartbeat loop without touching the heartbeat object — the |
| 282 | // shape a crashed box leaves behind (its record rots until staleAfter). The |
| 283 | // drain path is stop; tests use halt to simulate hard death. |
| 284 | func (m *meshSvc) halt() { |
| 285 | m.mu.Lock() |
| 286 | started := m.started |
| 287 | m.started = false |
| 288 | m.mu.Unlock() |
| 289 | if !started { |
| 290 | return |
| 291 | } |
| 292 | close(m.stopCh) |
| 293 | <-m.done |
| 294 | } |
| 295 | |
| 296 | // stop is the drain path: halt the loop (a beat racing the delete would |
| 297 | // resurrect us), then delete our heartbeat so peers take over immediately. |
| 298 | // The caller (activatorSvc.shutdown) checkpoints and releases AFTER this — |
| 299 | // that ordering is the drain contract. |
| 300 | func (m *meshSvc) stop(ctx context.Context) { |
| 301 | m.halt() |
| 302 | if err := m.store.Delete(ctx, m.self); err != nil { |
| 303 | log.Printf("mesh: heartbeat delete failed: %v — peers take over when it goes stale (%s)", err, m.staleAfter) |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // liveBoxes is the current membership: every fresh heartbeat with a sane peer |
| 308 | // URL, self always included — our own PUT may not have landed or listed yet, |
| 309 | // and a box always counts itself, so the set is never empty and hrwOwner is |
| 310 | // always defined. Staleness is applied at read time, so a box that dies |
| 311 | // between polls drops out the moment its record ages past staleAfter. |
| 312 | func (m *meshSvc) liveBoxes() []string { |
| 313 | now := m.now() |
| 314 | m.mu.Lock() |
| 315 | members := m.members |
| 316 | m.mu.Unlock() |
| 317 | boxes := []string{m.self} |
| 318 | for _, hb := range members { |
| 319 | if hb.Peer == m.self || now.Sub(hb.LastSeen) > m.staleAfter { |
| 320 | continue |
| 321 | } |
| 322 | // A garbage record (someone else's object decoding to nonsense) must |
| 323 | // not enter the hash — it would attract ~1/N of hosts and blackhole |
| 324 | // them behind an unproxyable "peer". |
| 325 | if u, err := url.Parse(hb.Peer); err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { |
| 326 | continue |
| 327 | } |
| 328 | boxes = append(boxes, hb.Peer) |
| 329 | } |
| 330 | return boxes |
| 331 | } |
| 332 | |
| 333 | // candidatesFor returns the live boxes eligible to own host — the member set |
| 334 | // hrwOwner picks from. Placement keys on the space's home_region (registry.go, |
| 335 | // replicated to every in-scope box by meshreplicate.go), so a space pinned to a |
| 336 | // region is owned by a live box IN that region; relocation is a home_region flip |
| 337 | // and the hash re-homes it onto the destination box (docs/multi-edge.md). |
| 338 | // |
| 339 | // The fallback to ALL live boxes is load-bearing: a space with no home_region |
| 340 | // (every space today), or one whose region has no live box right now, hashes |
| 341 | // across the whole fleet exactly as before regions existed. So with no |
| 342 | // LCHAT_MESH_REGION set anywhere and empty home_region everywhere, this returns |
| 343 | // liveBoxes() and owns/remoteOwner are byte-identical to pre-region placement. |
| 344 | func (m *meshSvc) candidatesFor(host string) []string { |
| 345 | live := m.liveBoxes() |
| 346 | home := m.homeRegionOf(host) |
| 347 | if home == "" { |
| 348 | return live // region-agnostic space — the whole fleet, as before regions |
| 349 | } |
| 350 | m.mu.Lock() |
| 351 | regions := m.regions |
| 352 | m.mu.Unlock() |
| 353 | var matched []string |
| 354 | for _, b := range live { |
| 355 | region := regions[b] |
| 356 | if b == m.self { |
| 357 | region = m.region // our own region is known locally, never via S3 |
| 358 | } |
| 359 | if region == home { |
| 360 | matched = append(matched, b) |
| 361 | } |
| 362 | } |
| 363 | if len(matched) == 0 { |
| 364 | // No live box advertises the space's region (none configured for it yet, |
| 365 | // or that region's box is down): fall back to the whole fleet so the |
| 366 | // space is still served somewhere — availability, and exact back-compat. |
| 367 | return live |
| 368 | } |
| 369 | return matched |
| 370 | } |
| 371 | |
| 372 | // homeRegionOf reads host's replicated home_region from the LOCAL registry |
| 373 | // (meshreplicate.go copies it onto every in-scope box, so no network). "" when |
| 374 | // the mesh has no registry handle (direct-newMesh unit tests), the host has no |
| 375 | // row, or its region is unset — all of which candidatesFor treats as agnostic. |
| 376 | func (m *meshSvc) homeRegionOf(host string) string { |
| 377 | if m.reg == nil { |
| 378 | return "" |
| 379 | } |
| 380 | r, err := m.reg.get(host) |
| 381 | if err != nil || r == nil { |
| 382 | return "" |
| 383 | } |
| 384 | return r.HomeRegion |
| 385 | } |
| 386 | |
| 387 | // regionLive reports whether any live box currently advertises region — the |
| 388 | // precondition `lchat relocate` checks before flipping a space's home_region |
| 389 | // onto a destination region: there is no point re-homing a space to a region |
| 390 | // no box serves (candidatesFor would just fall back to the whole fleet and the |
| 391 | // space would never actually move). It reads the same cached membership + |
| 392 | // peer-region view HRW placement uses, so the caller must beat() first to |
| 393 | // refresh it (fanOutMeshRoute's pattern). This box's own region is known |
| 394 | // locally (m.region), never via S3. |
| 395 | func (m *meshSvc) regionLive(region string) bool { |
| 396 | if region == "" { |
| 397 | return false |
| 398 | } |
| 399 | if m.region == region { |
| 400 | return true |
| 401 | } |
| 402 | m.mu.Lock() |
| 403 | regions := m.regions |
| 404 | m.mu.Unlock() |
| 405 | for _, b := range m.liveBoxes() { |
| 406 | if b == m.self { |
| 407 | continue // handled by m.region above |
| 408 | } |
| 409 | if regions[b] == region { |
| 410 | return true |
| 411 | } |
| 412 | } |
| 413 | return false |
| 414 | } |
| 415 | |
| 416 | // inScope reports whether host participates in HRW placement at all. Only |
| 417 | // hosts under a configured LCHAT_MESH_DOMAINS domain are ever hashed across |
| 418 | // boxes; everything else always resolves to whichever box is asked — correct |
| 419 | // because registerInstance only ever replicates a route to the boxes covered |
| 420 | // by those same domains (meshreplicate.go), so no other box has a row for it |
| 421 | // anyway, and this box (which was asked) is the only one that could possibly |
| 422 | // serve it. An empty domain list is unrestricted — a test-construction |
| 423 | // convenience (direct newMesh callers exercising bare HRW); meshFromEnv |
| 424 | // always supplies a non-empty list in production. |
| 425 | func (m *meshSvc) inScope(host string) bool { |
| 426 | return hostInDomains(host, m.domains) |
| 427 | } |
| 428 | |
| 429 | // hostInDomains is inScope's matching rule, factored out so `lchat mesh-seed` |
| 430 | // (meshreplicate.go) can filter routes to replicate without needing a full |
| 431 | // meshSvc: host equals a configured domain, or is a subdomain of one. |
| 432 | func hostInDomains(host string, domains []string) bool { |
| 433 | if len(domains) == 0 { |
| 434 | return true |
| 435 | } |
| 436 | host = strings.ToLower(host) |
| 437 | for _, d := range domains { |
| 438 | if host == d || strings.HasSuffix(host, "."+d) { |
| 439 | return true |
| 440 | } |
| 441 | } |
| 442 | return false |
| 443 | } |
| 444 | |
| 445 | // owns reports whether this box is host's HRW owner in the live member set — |
| 446 | // the due-time sweeper's gate: only the owner due-wakes a host. Out-of-scope |
| 447 | // hosts always report true: this box is the only one ever asked about them |
| 448 | // (see inScope), so there is nothing to arbitrate. |
| 449 | func (m *meshSvc) owns(host string) bool { |
| 450 | if !m.inScope(host) { |
| 451 | return true |
| 452 | } |
| 453 | return hrwOwner(host, m.candidatesFor(host)) == m.self |
| 454 | } |
| 455 | |
| 456 | // remoteOwner returns the peer to proxy host to when its HRW owner is another |
| 457 | // live box. ok=false means serve locally: host is out of mesh scope, we are |
| 458 | // the owner, or the previous owner is dead/stale (it's out of the member set, |
| 459 | // so the hash has already reassigned — possibly to us — and a local wake is |
| 460 | // the failover). |
| 461 | func (m *meshSvc) remoteOwner(host string) (peer string, ok bool) { |
| 462 | if !m.inScope(host) { |
| 463 | return "", false |
| 464 | } |
| 465 | if o := hrwOwner(host, m.candidatesFor(host)); o != m.self { |
| 466 | return o, true |
| 467 | } |
| 468 | return "", false |
| 469 | } |
| 470 | |
| 471 | // proxyTo forwards one request to the owning peer, instance hostname |
| 472 | // preserved. Proxies are built lazily, one per peer, and live for the mesh's |
| 473 | // life — peer URLs are stable box identities. |
| 474 | func (m *meshSvc) proxyTo(w http.ResponseWriter, r *http.Request, peer string) { |
| 475 | m.mu.Lock() |
| 476 | p := m.proxies[peer] |
| 477 | if p == nil { |
| 478 | u, err := url.Parse(peer) |
| 479 | if err != nil || u.Host == "" { |
| 480 | // liveBoxes filters garbage, so this is belt and braces. |
| 481 | m.mu.Unlock() |
| 482 | log.Printf("mesh: unusable peer URL %q for %s", peer, r.Host) |
| 483 | http.Error(w, "instance's owner box is unreachable; retry shortly", http.StatusServiceUnavailable) |
| 484 | return |
| 485 | } |
| 486 | p = newPeerProxy(m.self, u) |
| 487 | m.proxies[peer] = p |
| 488 | } |
| 489 | m.mu.Unlock() |
| 490 | p.ServeHTTP(w, r) |
| 491 | } |
| 492 | |
| 493 | // newPeerProxy builds the reverse proxy to one peer box. The peer routes by |
| 494 | // Host exactly like the public front door, so the instance hostname must |
| 495 | // survive the hop; the scheme comes from the peer URL and is honored: |
| 496 | // |
| 497 | // - http peers (a private network between boxes): dial the peer address, |
| 498 | // Host header = the instance host. |
| 499 | // - https peers: SNI *and* certificate verification run against the |
| 500 | // instance hostname — the peer presents its wildcard/per-host cert, which |
| 501 | // proves it serves this host (it could never prove it IS 10.x.y.z). The |
| 502 | // outgoing URL keeps the instance host and a custom dialer pins the TCP |
| 503 | // connection to the peer's address — the same shape as the box's own |
| 504 | // /etc/hosts loopback pin. InsecureSkipVerify is never used. |
| 505 | func newPeerProxy(self string, peer *url.URL) *httputil.ReverseProxy { |
| 506 | dialAddr := peerDialAddr(peer) |
| 507 | return &httputil.ReverseProxy{ |
| 508 | Rewrite: func(pr *httputil.ProxyRequest) { |
| 509 | pr.SetXForwarded() |
| 510 | pr.Out.URL.Scheme = peer.Scheme |
| 511 | if peer.Scheme == "https" { |
| 512 | pr.Out.URL.Host = pr.In.Host // SNI + cert verification target |
| 513 | } else { |
| 514 | pr.Out.URL.Host = peer.Host |
| 515 | } |
| 516 | pr.Out.Host = pr.In.Host // the peer routes by Host, like any client |
| 517 | pr.Out.Header.Set(meshForwardedHeader, self) |
| 518 | }, |
| 519 | Transport: &http.Transport{ |
| 520 | DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) { |
| 521 | return (&net.Dialer{Timeout: 10 * time.Second}).DialContext(ctx, network, dialAddr) |
| 522 | }, |
| 523 | // Bounded like unixProxy's pool, for the same leak reasons. |
| 524 | MaxIdleConns: 32, |
| 525 | IdleConnTimeout: 90 * time.Second, |
| 526 | }, |
| 527 | ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { |
| 528 | // An owner that died but hasn't gone stale yet (or a box-to-box |
| 529 | // network fault): 503 + retry. Within staleAfter the membership |
| 530 | // view drops it and retries wake the instance locally instead — |
| 531 | // staleness bounds this window. |
| 532 | log.Printf("mesh: proxy %s to %s: %v", r.Host, peer, err) |
| 533 | w.Header().Set("Retry-After", "5") |
| 534 | http.Error(w, "instance's owner box is unreachable; retry shortly", http.StatusServiceUnavailable) |
| 535 | }, |
| 536 | } |
| 537 | } |
| 538 | |
| 539 | // peerDialAddr is the TCP address a peer base URL dials to, default ports |
| 540 | // applied — computed once per proxy, since the peer URL never changes. |
| 541 | func peerDialAddr(peer *url.URL) string { |
| 542 | if peer.Port() != "" { |
| 543 | return peer.Host |
| 544 | } |
| 545 | if peer.Scheme == "https" { |
| 546 | return peer.Hostname() + ":443" |
| 547 | } |
| 548 | return peer.Hostname() + ":80" |
| 549 | } |
| 550 | |
| 551 | // hrwOwner picks host's owner from boxes by rendezvous (highest-random-weight) |
| 552 | // hashing: the box with the highest hash(box||host) wins. A pure function of |
| 553 | // (host, member set) — every box computes the same answer from membership |
| 554 | // alone, and when a box joins or leaves only ~1/N of hosts change owner (a |
| 555 | // joiner only steals, a leaver's hosts scatter over the survivors; nothing |
| 556 | // else moves). The result is a placement HINT, not a lock: the S3 lease stays |
| 557 | // the single-writer truth, which is why this can be twenty lines. |
| 558 | func hrwOwner(host string, boxes []string) string { |
| 559 | var best string |
| 560 | var bestScore uint64 |
| 561 | for _, b := range boxes { |
| 562 | h := fnv.New64a() |
| 563 | io.WriteString(h, b) |
| 564 | h.Write([]byte{0}) // separator: ("ab","c") must not collide with ("a","bc") |
| 565 | io.WriteString(h, host) |
| 566 | if s := h.Sum64(); best == "" || s > bestScore || (s == bestScore && b > best) { |
| 567 | best, bestScore = b, s |
| 568 | } |
| 569 | } |
| 570 | return best |
| 571 | } |
| 572 | |
amadan