| 1 | // Package relay: serve.go is the process. Everything else in this package |
| 2 | // binds a socket or parses a record; this is the loop that owns them for the |
| 3 | // life of `carlos relay serve` (spec §3). It reads the masters file as it |
| 4 | // changes, the authority record every minute and on every control/serial |
| 5 | // change, writes a heartbeat on each read, flushes the meter each minute, |
| 6 | // sweeps expiry, and sheds under the box's memory pressure. |
| 7 | package relay |
| 8 | |
| 9 | import ( |
| 10 | "bufio" |
| 11 | "context" |
| 12 | "errors" |
| 13 | "fmt" |
| 14 | "net" |
| 15 | "os" |
| 16 | "slices" |
| 17 | "strconv" |
| 18 | "strings" |
| 19 | "sync" |
| 20 | "time" |
| 21 | |
| 22 | "github.com/carlosframework/platform/internal/release" |
| 23 | ) |
| 24 | |
| 25 | // The relay's fixed listeners (spec §3). Ports, not addresses: 3478 is the |
| 26 | // registered STUN/TURN port and the DNS names and security group are |
| 27 | // written against it, so it is not tunable. |
| 28 | const ( |
| 29 | relayUDPAddr = ":3478" |
| 30 | relayTCPAddr = ":3478" |
| 31 | |
| 32 | // defaultMastersPath is where the operator stages the region masters |
| 33 | // and what carlos-relay.service's ConditionPathExists names. |
| 34 | defaultMastersPath = "/etc/carlos/relay-masters" |
| 35 | |
| 36 | // defaultPressurePath is the kernel's memory pressure-stall file. The |
| 37 | // relay reads the `some avg10` value out of it (spec §3). |
| 38 | defaultPressurePath = "/proc/pressure/memory" |
| 39 | |
| 40 | // shedAbove and shedUntil are the pressure hysteresis from spec §3: the |
| 41 | // relay starts shedding above 20% and keeps going until pressure falls |
| 42 | // under 10%, so a relay that has just freed a little memory does not |
| 43 | // flap in and out of shedding on every tick. |
| 44 | shedAbove = 20.0 |
| 45 | shedUntil = 10.0 |
| 46 | ) |
| 47 | |
| 48 | // The serve loop's cadences. Vars, not consts, so a test can shorten them. |
| 49 | var ( |
| 50 | authInterval = time.Minute |
| 51 | serialInterval = 3 * time.Second |
| 52 | meterInterval = time.Minute |
| 53 | expiryInterval = 10 * time.Second |
| 54 | psiInterval = 5 * time.Second |
| 55 | |
| 56 | // The store-open backoff (OpenStoreWithRetry). |
| 57 | storeRetryInitial = time.Second |
| 58 | storeRetryMax = 30 * time.Second |
| 59 | storeRetryLogEvery = time.Minute |
| 60 | ) |
| 61 | |
| 62 | // OpenStoreWithRetry calls open until it succeeds or ctx is done, backing |
| 63 | // off from a second to half a minute and logging at most one line a minute. |
| 64 | // |
| 65 | // It exists because the alternative is worse than it looks. The relay's |
| 66 | // unit is Restart=always, and systemd's default start limit turns five |
| 67 | // restarts in ten seconds into a unit that sits in `failed` and stays |
| 68 | // there: a box whose deployment bucket was unreachable for the fifteen |
| 69 | // seconds around boot would then serve no STUN until a platform update or |
| 70 | // a human touched it — and STUN, the whole reason this stage exists, needs |
| 71 | // no store at all. A relay that waits is a relay that comes back on its |
| 72 | // own. Configuration errors are NOT retried here; they are fatal at the |
| 73 | // call site, because no amount of waiting fixes a missing public IP. |
| 74 | func OpenStoreWithRetry(ctx context.Context, open func(context.Context) (release.Store, error), logf func(string, ...any)) (release.Store, error) { |
| 75 | if logf == nil { |
| 76 | logf = func(string, ...any) {} |
| 77 | } |
| 78 | delay := storeRetryInitial |
| 79 | var lastLog time.Time |
| 80 | for attempt := 1; ; attempt++ { |
| 81 | s, err := open(ctx) |
| 82 | if err == nil { |
| 83 | if attempt > 1 { |
| 84 | logf("relay: deploy store opened on attempt %d", attempt) |
| 85 | } |
| 86 | return s, nil |
| 87 | } |
| 88 | if now := time.Now(); lastLog.IsZero() || now.Sub(lastLog) >= storeRetryLogEvery { |
| 89 | logf("relay: deploy store unavailable after %d attempt(s): %v (retrying in %s; STUN is unaffected)", attempt, err, delay) |
| 90 | lastLog = now |
| 91 | } |
| 92 | select { |
| 93 | case <-ctx.Done(): |
| 94 | return nil, ctx.Err() |
| 95 | case <-time.After(delay): |
| 96 | } |
| 97 | delay *= 2 |
| 98 | if delay > storeRetryMax { |
| 99 | delay = storeRetryMax |
| 100 | } |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | // Config is everything `carlos relay serve` needs from its box: who it is, |
| 105 | // where it listens from, what it may not forward to, and the store it reads |
| 106 | // its authority from and writes its heartbeat and metering to. |
| 107 | type Config struct { |
| 108 | Box string // the metrics box label; the key its heartbeat and meter objects live under |
| 109 | Region string // the AWS region name, as the edges file spells it |
| 110 | |
| 111 | MastersPath string // /etc/carlos/relay-masters |
| 112 | PublicIP net.IP // the EIP advertised to clients as the relay address |
| 113 | |
| 114 | DenyCIDRs []*net.IPNet // peer ranges refused on top of the built-in private list |
| 115 | Store release.Store |
| 116 | Limits Limits |
| 117 | |
| 118 | // PressurePath is /proc/pressure/memory. A path that cannot be read |
| 119 | // (an old kernel, a container without PSI) simply disables shedding — |
| 120 | // the cgroup caps in the unit are still there — rather than failing. |
| 121 | PressurePath string |
| 122 | |
| 123 | // udp and tcp are the listen addresses, unexported because they are a |
| 124 | // test seam and nothing else: production is :3478 on both, per spec §3, |
| 125 | // and no env may move them. Empty means the constants above. |
| 126 | udp, tcp string |
| 127 | |
| 128 | // extraAllow is the same kind of seam, for the same reason. A relay on a |
| 129 | // real box never permits a loopback peer, and an end-to-end test that |
| 130 | // carries traffic between two sockets on this machine has nowhere else |
| 131 | // to put them. Unexported so no env and no operator can widen the peer |
| 132 | // policy; ServerConfig.ExtraAllow is the reviewed, production-facing one. |
| 133 | extraAllow []*net.IPNet |
| 134 | } |
| 135 | |
| 136 | // ConfigFromEnv builds a Config from the unit's environment. imdsIP and |
| 137 | // imdsRegion are the IMDS probes (host.ReadPublicIP, host.ReadPlacementRegion), |
| 138 | // consulted only when the matching override is unset, so a box off EC2 is |
| 139 | // configured entirely by host.env. |
| 140 | // |
| 141 | // Box is deliberately NOT read from IMDS or derived here: the label must be |
| 142 | // the one the metrics loop publishes under (host.BoxLabel), which lives in |
| 143 | // internal/host, so the caller fills it in when CARLOS_RELAY_BOX is unset. |
| 144 | func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error) { |
| 145 | _ = ctx // the probes are the caller's; nothing here does I/O of its own |
| 146 | if getenv == nil { |
| 147 | getenv = func(string) string { return "" } |
| 148 | } |
| 149 | env := func(name string) string { return strings.TrimSpace(getenv(name)) } |
| 150 | |
| 151 | cfg := Config{ |
| 152 | Box: env("CARLOS_RELAY_BOX"), |
| 153 | MastersPath: defaultMastersPath, |
| 154 | PressurePath: defaultPressurePath, |
| 155 | Store: store, |
| 156 | Limits: LimitsFromEnv(getenv), |
| 157 | } |
| 158 | if p := env("CARLOS_RELAY_MASTERS_PATH"); p != "" { |
| 159 | cfg.MastersPath = p |
| 160 | } |
| 161 | if p := env("CARLOS_RELAY_PRESSURE_PATH"); p != "" { |
| 162 | cfg.PressurePath = p |
| 163 | } |
| 164 | |
| 165 | // The region rule is edgesFileRegion's, spelled again here rather than |
| 166 | // imported, because the two must agree: CARLOS_AWS_REGION wins outright, |
| 167 | // otherwise the IMDS placement probe stands. Empty is not fatal — such a |
| 168 | // relay matches no app's region list and serves STUN only, which is a |
| 169 | // legitimate state, not a misconfiguration to die on. |
| 170 | cfg.Region = env("CARLOS_AWS_REGION") |
| 171 | if cfg.Region == "" && imdsRegion != nil { |
| 172 | cfg.Region = strings.TrimSpace(imdsRegion()) |
| 173 | } |
| 174 | |
| 175 | ip := env("CARLOS_RELAY_PUBLIC_IP") |
| 176 | if ip == "" && imdsIP != nil { |
| 177 | ip = strings.TrimSpace(imdsIP()) |
| 178 | } |
| 179 | if ip == "" { |
| 180 | return Config{}, errors.New("relay: no public IP: set CARLOS_RELAY_PUBLIC_IP, or run on a box whose IMDS reports a public-ipv4") |
| 181 | } |
| 182 | cfg.PublicIP = net.ParseIP(ip) |
| 183 | if cfg.PublicIP == nil { |
| 184 | return Config{}, fmt.Errorf("relay: CARLOS_RELAY_PUBLIC_IP %q is not an IP address", ip) |
| 185 | } |
| 186 | |
| 187 | // A typo in the deny list must stop the relay, not quietly open the VPC |
| 188 | // to any browser that can reach 3478. |
| 189 | for _, s := range strings.Split(env("CARLOS_RELAY_DENY_CIDRS"), ",") { |
| 190 | s = strings.TrimSpace(s) |
| 191 | if s == "" { |
| 192 | continue |
| 193 | } |
| 194 | _, n, err := net.ParseCIDR(s) |
| 195 | if err != nil { |
| 196 | return Config{}, fmt.Errorf("relay: CARLOS_RELAY_DENY_CIDRS %q: %w", s, err) |
| 197 | } |
| 198 | cfg.DenyCIDRs = append(cfg.DenyCIDRs, n) |
| 199 | } |
| 200 | return cfg, nil |
| 201 | } |
| 202 | |
| 203 | // Serve runs the relay until ctx is done, then closes every listener and |
| 204 | // allocation and returns nil. The only errors it returns are from starting: |
| 205 | // once it is listening, every failure is logged and retried on the next |
| 206 | // tick, because a relay that exits on a bad bucket read is a relay that |
| 207 | // stops answering STUN for a reason that has nothing to do with STUN. |
| 208 | func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error { |
| 209 | if logf == nil { |
| 210 | logf = func(string, ...any) {} |
| 211 | } |
| 212 | switch { |
| 213 | case cfg.Box == "": |
| 214 | return errors.New("relay: Config.Box is required (the metrics box label)") |
| 215 | case cfg.Store == nil: |
| 216 | return errors.New("relay: Config.Store is required") |
| 217 | case cfg.PublicIP == nil: |
| 218 | return errors.New("relay: Config.PublicIP is required") |
| 219 | } |
| 220 | if cfg.MastersPath == "" { |
| 221 | cfg.MastersPath = defaultMastersPath |
| 222 | } |
| 223 | if cfg.Region == "" { |
| 224 | logf("relay: no region: this box matches no app's region list and will serve STUN only") |
| 225 | } |
| 226 | |
| 227 | // The four live views the server reads on every request. They are |
| 228 | // written here, under mu, and read on pion's goroutines, so every |
| 229 | // accessor takes the lock and hands back a value the server may hold. |
| 230 | var ( |
| 231 | mu sync.Mutex |
| 232 | masters *Masters |
| 233 | mastersE string |
| 234 | auth *Authority |
| 235 | lease Lease |
| 236 | ) |
| 237 | now := time.Now |
| 238 | meter := NewMeter(cfg.Box, now, logf) |
| 239 | |
| 240 | // Each defaults on its own, so an empty tcp seam can never silently |
| 241 | // take the UDP listener down with it (or the reverse): spec §3 says |
| 242 | // both listeners, always, and a half-configured pair is a bug rather |
| 243 | // than a supported shape. A test seams both or neither. |
| 244 | udp, tcp := cfg.udp, cfg.tcp |
| 245 | if udp == "" { |
| 246 | udp = relayUDPAddr |
| 247 | } |
| 248 | if tcp == "" { |
| 249 | tcp = relayTCPAddr |
| 250 | } |
| 251 | srv, err := NewServer(ServerConfig{ |
| 252 | Region: cfg.Region, |
| 253 | RelayIP: cfg.PublicIP, |
| 254 | UDP: udp, |
| 255 | TCP: tcp, |
| 256 | MinPort: 49152, |
| 257 | MaxPort: 65535, |
| 258 | DenyCIDRs: cfg.DenyCIDRs, |
| 259 | ExtraAllow: cfg.extraAllow, |
| 260 | Limits: cfg.Limits, |
| 261 | Now: now, |
| 262 | Log: logf, |
| 263 | Meter: meter, |
| 264 | Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters }, |
| 265 | Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth }, |
| 266 | Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) }, |
| 267 | }) |
| 268 | if err != nil { |
| 269 | return err |
| 270 | } |
| 271 | defer srv.Close() |
| 272 | logf("relay: listening on %v, relay address %s, region %q", srv.Listeners(), cfg.PublicIP, cfg.Region) |
| 273 | |
| 274 | // Rebuild the rolling-hour byte windows from what this box already |
| 275 | // metered this hour and last (spec §3, "Limits"): a restart must not |
| 276 | // hand an app that has spent its 20 GB a fresh budget. |
| 277 | if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now(), logf); err != nil { |
| 278 | logf("relay: rolling-hour windows not rebuilt (%v); budgets start empty", err) |
| 279 | } else { |
| 280 | seedWindows(srv, samples) |
| 281 | } |
| 282 | |
| 283 | go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) { |
| 284 | mu.Lock() |
| 285 | if err != nil { |
| 286 | mastersE = err.Error() |
| 287 | } else { |
| 288 | masters, mastersE = m, "" |
| 289 | } |
| 290 | mu.Unlock() |
| 291 | if err != nil { |
| 292 | logf("relay: masters file %s: %v", cfg.MastersPath, err) |
| 293 | return |
| 294 | } |
| 295 | // A reload can retire a master, and an allocation signed by it must |
| 296 | // end now rather than at the next expiry sweep. |
| 297 | srv.tick() |
| 298 | }) |
| 299 | |
| 300 | // readAuth is one authority read: it renews the lease on success only |
| 301 | // (a failed read renews nothing, spec §3), ends what the new record no |
| 302 | // longer allows, and writes the heartbeat that tells the console this |
| 303 | // box is current. |
| 304 | readAuth := func() { |
| 305 | a, _, err := ReadAuthority(ctx, cfg.Store) |
| 306 | t := now() |
| 307 | listeners := srv.Listeners() // outside mu: it takes the server's own lock |
| 308 | mu.Lock() |
| 309 | if err == nil { |
| 310 | auth = a |
| 311 | lease.Renew(t) |
| 312 | } |
| 313 | cur, live := auth, lease.Live(t) |
| 314 | hb := Heartbeat{ |
| 315 | Box: cfg.Box, |
| 316 | Region: cfg.Region, |
| 317 | ReadAt: lease.Last(), // the last SUCCESSFUL read, never the wall clock |
| 318 | Listeners: listeners, |
| 319 | Kids: kidsOf(masters), |
| 320 | Retiring: retiringOf(masters), |
| 321 | MastersError: mastersE, |
| 322 | } |
| 323 | if err != nil { |
| 324 | hb.AuthorityError = err.Error() |
| 325 | } |
| 326 | if cur != nil { |
| 327 | hb.WrittenAt = cur.WrittenAt |
| 328 | } |
| 329 | mu.Unlock() |
| 330 | if err != nil { |
| 331 | logf("relay: authority read: %v", err) |
| 332 | } |
| 333 | if !live { |
| 334 | srv.EndAll("lease lapsed") |
| 335 | } |
| 336 | // cur is a snapshot, so the predicate runs off the lock without |
| 337 | // racing the next read. A nil record means no app is enabled here |
| 338 | // and every allocation goes, which is the "STUN only" state. |
| 339 | srv.EndWhere(func(b *Binding) bool { |
| 340 | x, ok := cur.Lookup(b.ID.Account, b.ID.App) |
| 341 | return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region) |
| 342 | }, "authority") |
| 343 | if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil { |
| 344 | logf("relay: heartbeat: %v", err) |
| 345 | } |
| 346 | } |
| 347 | readAuth() |
| 348 | |
| 349 | flush := func(ctx context.Context) { |
| 350 | // A complete census, not one id at a time: an app whose last |
| 351 | // allocation ended has to be told its gauge is zero, or it reports |
| 352 | // a phantom allocation forever and the meter never evicts it. |
| 353 | meter.SetAllocsAll(srv.LiveByIdentity()) |
| 354 | if err := meter.Flush(ctx, cfg.Store); err != nil { |
| 355 | logf("relay: meter flush: %v", err) |
| 356 | } |
| 357 | } |
| 358 | |
| 359 | authTick := time.NewTicker(authInterval) |
| 360 | serialTick := time.NewTicker(serialInterval) |
| 361 | meterTick := time.NewTicker(meterInterval) |
| 362 | expiryTick := time.NewTicker(expiryInterval) |
| 363 | psiTick := time.NewTicker(psiInterval) |
| 364 | defer func() { |
| 365 | for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} { |
| 366 | t.Stop() |
| 367 | } |
| 368 | }() |
| 369 | |
| 370 | var ( |
| 371 | lastSerial int64 = -1 |
| 372 | shedding bool |
| 373 | ) |
| 374 | for { |
| 375 | select { |
| 376 | case <-ctx.Done(): |
| 377 | // The last minute of metering is real money; give it its own |
| 378 | // short deadline rather than losing it to the cancelled ctx. |
| 379 | final, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) |
| 380 | flush(final) |
| 381 | cancel() |
| 382 | return nil |
| 383 | case <-authTick.C: |
| 384 | readAuth() |
| 385 | case <-serialTick.C: |
| 386 | // The relay polls control/serial itself: the agent's |
| 387 | // fastConvergeLoop is a different process (spec §3). |
| 388 | doc, _, err := release.ReadSerial(ctx, cfg.Store) |
| 389 | switch { |
| 390 | case err != nil: |
| 391 | logf("relay: serial read: %v", err) |
| 392 | case doc.Serial != lastSerial: |
| 393 | lastSerial = doc.Serial |
| 394 | readAuth() |
| 395 | } |
| 396 | case <-meterTick.C: |
| 397 | flush(ctx) |
| 398 | case <-expiryTick.C: |
| 399 | srv.tick() |
| 400 | case <-psiTick.C: |
| 401 | some, err := readPSI(cfg.PressurePath) |
| 402 | if err != nil { |
| 403 | continue |
| 404 | } |
| 405 | // Spec §3 asks for shedding above 20% and a resumption of |
| 406 | // admission under 10%. The Server has no admission pause to |
| 407 | // lift — admission is governed by the limits, not by pressure — |
| 408 | // so "resume" here is simply the end of shedding: once over 20% |
| 409 | // the relay drops its oldest allocation every tick until |
| 410 | // pressure falls under 10%, and then stops. |
| 411 | switch { |
| 412 | case some > shedAbove: |
| 413 | shedding = true |
| 414 | case some < shedUntil: |
| 415 | shedding = false |
| 416 | } |
| 417 | if shedding { |
| 418 | srv.ShedOldest(1) |
| 419 | } |
| 420 | } |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | // seedWindows replays a window of metered samples into the admission |
| 425 | // counters. A sample's app key is "<account>/<app>" and an app name can |
| 426 | // never contain a slash (spec §2's grammar), so the first slash splits it. |
| 427 | func seedWindows(srv *Server, samples []MeterSample) { |
| 428 | for _, s := range samples { |
| 429 | for key, app := range s.Apps { |
| 430 | account, name, ok := strings.Cut(key, "/") |
| 431 | if !ok || account == "" || name == "" { |
| 432 | continue |
| 433 | } |
| 434 | srv.Seed(Identity{Account: account, App: name}, s.At, app.In+app.Out) |
| 435 | } |
| 436 | } |
| 437 | } |
| 438 | |
| 439 | // kidsOf is Masters.Kids with a nil guard: the heartbeat reports the kids |
| 440 | // currently loaded, and "none, because the file has never parsed" is a |
| 441 | // state the console wants to see rather than a panic. |
| 442 | func kidsOf(m *Masters) []string { |
| 443 | if m == nil { |
| 444 | return nil |
| 445 | } |
| 446 | return m.Kids() |
| 447 | } |
| 448 | |
| 449 | // retiringOf is Masters.Retiring with the same nil guard kidsOf carries. |
| 450 | func retiringOf(m *Masters) map[string]time.Time { |
| 451 | if m == nil { |
| 452 | return nil |
| 453 | } |
| 454 | r := m.Retiring() |
| 455 | if len(r) == 0 { |
| 456 | return nil |
| 457 | } |
| 458 | return r |
| 459 | } |
| 460 | |
| 461 | // readPSI reads the `some avg10` percentage out of a pressure-stall file |
| 462 | // (/proc/pressure/memory): |
| 463 | // |
| 464 | // some avg10=0.00 avg60=0.00 avg300=0.00 total=0 |
| 465 | // full avg10=0.00 avg60=0.00 avg300=0.00 total=0 |
| 466 | // |
| 467 | // Only the `some` line is used: it is the share of time ANY task stalled on |
| 468 | // memory, which is what "the box is under pressure" means for a unit that |
| 469 | // has to lose first. Plain file reading, so this compiles on the platforms |
| 470 | // the CLI cross-builds for and simply errors there. |
| 471 | func readPSI(path string) (float64, error) { |
| 472 | f, err := os.Open(path) |
| 473 | if err != nil { |
| 474 | return 0, err |
| 475 | } |
| 476 | defer f.Close() |
| 477 | sc := bufio.NewScanner(f) |
| 478 | for sc.Scan() { |
| 479 | fields := strings.Fields(sc.Text()) |
| 480 | if len(fields) == 0 || fields[0] != "some" { |
| 481 | continue |
| 482 | } |
| 483 | for _, kv := range fields[1:] { |
| 484 | v, ok := strings.CutPrefix(kv, "avg10=") |
| 485 | if !ok { |
| 486 | continue |
| 487 | } |
| 488 | n, err := strconv.ParseFloat(v, 64) |
| 489 | if err != nil { |
| 490 | return 0, fmt.Errorf("%s: some avg10=%q: %w", path, v, err) |
| 491 | } |
| 492 | return n, nil |
| 493 | } |
| 494 | } |
| 495 | if err := sc.Err(); err != nil { |
| 496 | return 0, err |
| 497 | } |
| 498 | return 0, fmt.Errorf("%s: no `some avg10=` field", path) |
| 499 | } |
| 500 | |
amadan