amadan

carlos / platform Public

relay: carlos relay serve and keygen (spec §3)

Serve is the process the unit runs: it binds UDP and TCP 3478, watches
the masters file, reads control/relay.json every minute and on every
control/serial change it sees, writes the heartbeat the console reads,
flushes the meter each minute, sweeps expiry, and sheds under memory
pressure.

Three decisions worth naming.

The rolling-hour byte windows are rebuilt at start from the metrics
objects this box wrote (LoadWindow), so a restart does not hand an app
that has spent its 20 GB a fresh budget.

The pressure rule reads /proc/pressure/memory's `some avg10` and holds
the spec's 20/10 hysteresis as one state: over 20% the relay drops its
oldest allocation every five seconds until pressure falls under 10%.
There is no admission to "resume" — admission is governed by the limits,
not by pressure — so resuming is simply the end of shedding.

ConfigFromEnv applies CARLOS_AWS_REGION exactly as edgesFileRegion does,
because the relay's region and the edges file's must be the same string,
and refuses to start with no public IP: a relay that cannot advertise a
relay address can allocate nothing. A malformed CARLOS_RELAY_DENY_CIDRS
also refuses, rather than quietly opening the VPC to anything that can
reach 3478.

The box label comes from host.BoxLabel() in the verb, not from the relay
package, so the relay's heartbeat and hourly metering land under the same
name the metrics loop publishes under. ReadPublicIP, ReadPlacementRegion,
OpenDeployStore and BoxLabel are thin exported wrappers over the existing
host functions; no logic moved.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 36faac6895ca5bcee2871b0b7e3f10da1f14c796 parent e5ecdac
9 files changed, +869 −2
  • docs/site/cli.md +25 −0
  • internal/host/agent.go +20 −0
  • internal/host/metrics_loop.go +7 −0
  • internal/relay/masters.go +6 −2
  • internal/relay/serve.go +420 −0
  • internal/relay/serve_test.go +276 −0
  • internal/relay/server.go +15 −0
  • main.go +4 −0
  • relay.go +96 −0
diff --git a/docs/site/cli.md b/docs/site/cli.md
index 9b13092e1..8e630b69c 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -664,6 +664,31 @@ loud failure beats a pass that silently finds nothing forever.
Wiring an engine is not the same as switching scanning on. The deployment-wide
switch is `carlos system antivirus` — see [carlos system](#carlos-system).
+### carlos relay
+
+The platform's STUN and TURN relay — the thing that lets a browser-to-browser
+call connect across the internet rather than only across a LAN. Two verbs
+exist so far, and neither is one an app developer types.
+
+```sh
+carlos relay keygen
+```
+
+`keygen` prints one line — a key id, a key, and `-` for "no retirement date"
+— for a region's master key. Stage it in `/etc/carlos/relay-masters` on that
+region's boxes, mode 0640, root-owned, group `carlos`, and put the same pair
+in the console's `CARLOS_RELAY_MASTERS_<REGION>` secret. The key is printed
+once and stored nowhere by this command.
+
+`serve` is what `carlos-relay.service` execs. The unit only starts on a box
+that has a masters file, so a box with no key never listens; a box with one
+answers STUN to anyone and TURN to credentials an enabled app minted. It
+reads its configuration from `/etc/carlos/host.env`:
+`CARLOS_RELAY_PUBLIC_IP` (the address it advertises, when IMDS cannot say),
+`CARLOS_RELAY_DENY_CIDRS` (peer ranges it refuses, on top of the private
+ones it always refuses), `CARLOS_AWS_REGION`, `CARLOS_RELAY_MASTERS_PATH`,
+and the `CARLOS_RELAY_LIMIT_*` family.
+
### carlos email
Tenant email sending. `carlos email enable` is the whole path in one
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 0208f0dc7..a877cb113 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -1821,6 +1821,12 @@ func readPublicIP() string {
return ip
}
+// ReadPublicIP is readPublicIP for callers outside this package — the relay
+// advertises the box's Elastic IP as its relay address and learns it exactly
+// as the edge does (relay spec §3, "Listeners"). Same contract: "" means
+// unknown, never an error.
+func ReadPublicIP() string { return readPublicIP() }
+
// readPlacementRegion best-effort reads this box's AWS region from EC2
// IMDSv2 (token PUT, then the placement/region leaf) — the same idiom and
// the same "" on any failure contract as readPublicIP: not on EC2, IMDS
@@ -1841,6 +1847,13 @@ func readPlacementRegion() string {
return region
}
+// ReadPlacementRegion is readPlacementRegion for callers outside this
+// package. The relay names its region in every heartbeat and checks it
+// against each app's region list, and it must be the SAME string the edges
+// file uses — hence the same probe, with the caller applying the same
+// CARLOS_AWS_REGION override edgesFileRegion applies.
+func ReadPlacementRegion() string { return readPlacementRegion() }
+
// domainsLoop runs the domains reconcile pass forever on the STATUS tick's
// cadence (5min + jitter), not the adopt loop's hour: the console's
// custom-domain guidance promises "Checks run every few minutes", and a DNS
@@ -2693,6 +2706,13 @@ func openDeployStore(ctx context.Context) (release.Store, error) {
return openDeployStoreEnv(ctx, os.Getenv)
}
+// OpenDeployStore is openDeployStore for callers outside this package. The
+// relay reads its authority record and writes its heartbeat and metering to
+// the same store the agent adopts from, chosen by the same rule, so a box
+// that has moved onto the fleet door does not leave the relay behind on a
+// bucket it can no longer reach.
+func OpenDeployStore(ctx context.Context) (release.Store, error) { return openDeployStore(ctx) }
+
// openDeployStoreEnv is openDeployStore with the environment injected — the
// same getenv seam resolveStatusBucket already uses in this file, so the
// selection rule can be tested without mutating process-wide env.
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go
index f3ffc1296..58a88aadf 100644
--- a/internal/host/metrics_loop.go
+++ b/internal/host/metrics_loop.go
@@ -739,6 +739,13 @@ func metricsBoxLabel(getenv func(string) string, machineID string) string {
return status.Label(machineID)
}
+// BoxLabel is metricsBoxLabel for callers outside this package — the relay
+// (`carlos relay serve`, relay spec §3) meters and heartbeats under the same
+// box label the metrics loop publishes under, so the two agree about which
+// box a row belongs to. A thin wrapper deliberately: the selection rule
+// stays in one place, tested there.
+func BoxLabel() string { return metricsBoxLabel(os.Getenv, readMachineID()) }
+
// metricsLoop runs the metrics sampler's 60s tick forever, once armed — thin
// and untested by the same convention as statusLoop/adoptLoop: everything
// with behavior lives in sampler.tick, metricsFlush, alertTick, and
diff --git a/internal/relay/masters.go b/internal/relay/masters.go
index bf6f1c695..c8c1f7431 100644
--- a/internal/relay/masters.go
+++ b/internal/relay/masters.go
@@ -12,6 +12,10 @@ import (
"time"
)
+// MasterKeyLen is the length of a region master in bytes (spec §4). Both
+// the parser and `carlos relay keygen` are held to it from here.
+const MasterKeyLen = 32
+
type Master struct {
Kid string
Key []byte
@@ -40,8 +44,8 @@ func ParseMasters(r io.Reader) (*Masters, error) {
return nil, fmt.Errorf("relay masters line %d: bad kid %q", line, f[0])
}
key, err := base64.StdEncoding.DecodeString(f[1])
- if err != nil || len(key) != 32 {
- return nil, fmt.Errorf("relay masters line %d: key must be base64 of 32 bytes", line)
+ if err != nil || len(key) != MasterKeyLen {
+ return nil, fmt.Errorf("relay masters line %d: key must be base64 of %d bytes", line, MasterKeyLen)
}
var retire time.Time
if f[2] != "-" {
diff --git a/internal/relay/serve.go b/internal/relay/serve.go
new file mode 100644
index 000000000..bd4e82201
--- /dev/null
+++ b/internal/relay/serve.go
@@ -0,0 +1,420 @@
+// Package relay: serve.go is the process. Everything else in this package
+// binds a socket or parses a record; this is the loop that owns them for
+// the life of
+// `carlos relay serve` (spec §3). It reads the masters file as it changes,
+// the authority record every minute and on every control/serial change,
+// writes a heartbeat on each read, flushes the meter each minute, sweeps
+// expiry, and sheds under the box's memory pressure.
+package relay
+
+import (
+ "bufio"
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "os"
+ "slices"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The relay's fixed listeners (spec §3). Ports, not addresses: 3478 is the
+// registered STUN/TURN port and the DNS names and security group are
+// written against it, so it is not tunable.
+const (
+ relayUDPAddr = ":3478"
+ relayTCPAddr = ":3478"
+
+ // defaultMastersPath is where the operator stages the region masters
+ // and what carlos-relay.service's ConditionPathExists names.
+ defaultMastersPath = "/etc/carlos/relay-masters"
+
+ // defaultPressurePath is the kernel's memory pressure-stall file. The
+ // relay reads the `some avg10` value out of it (spec §3).
+ defaultPressurePath = "/proc/pressure/memory"
+
+ // shedAbove and shedUntil are the pressure hysteresis from spec §3: the
+ // relay starts shedding above 20% and keeps going until pressure falls
+ // under 10%, so a relay that has just freed a little memory does not
+ // flap in and out of shedding on every tick.
+ shedAbove = 20.0
+ shedUntil = 10.0
+)
+
+// The serve loop's cadences. Vars, not consts, so a test can shorten them.
+var (
+ authInterval = time.Minute
+ serialInterval = 3 * time.Second
+ meterInterval = time.Minute
+ expiryInterval = 10 * time.Second
+ psiInterval = 5 * time.Second
+)
+
+// Config is everything `carlos relay serve` needs from its box: who it is,
+// where it listens from, what it may not forward to, and the store it reads
+// its authority from and writes its heartbeat and metering to.
+type Config struct {
+ Box string // the metrics box label; the key its heartbeat and meter objects live under
+ Region string // the AWS region name, as the edges file spells it
+
+ MastersPath string // /etc/carlos/relay-masters
+ PublicIP net.IP // the EIP advertised to clients as the relay address
+
+ DenyCIDRs []*net.IPNet // peer ranges refused on top of the built-in private list
+ Store release.Store
+ Limits Limits
+
+ // PressurePath is /proc/pressure/memory. A path that cannot be read
+ // (an old kernel, a container without PSI) simply disables shedding —
+ // the cgroup caps in the unit are still there — rather than failing.
+ PressurePath string
+
+ // udp and tcp are the listen addresses, unexported because they are a
+ // test seam and nothing else: production is :3478 on both, per spec §3,
+ // and no env may move them. Empty means the constants above.
+ udp, tcp string
+}
+
+// ConfigFromEnv builds a Config from the unit's environment. imdsIP and
+// imdsRegion are the IMDS probes (host.ReadPublicIP, host.ReadPlacementRegion),
+// consulted only when the matching override is unset, so a box off EC2 is
+// configured entirely by host.env.
+//
+// Box is deliberately NOT read from IMDS or derived here: the label must be
+// the one the metrics loop publishes under (host.BoxLabel), which lives in
+// internal/host, so the caller fills it in when CARLOS_RELAY_BOX is unset.
+func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error) {
+ _ = ctx // the probes are the caller's; nothing here does I/O of its own
+ if getenv == nil {
+ getenv = func(string) string { return "" }
+ }
+ env := func(name string) string { return strings.TrimSpace(getenv(name)) }
+
+ cfg := Config{
+ Box: env("CARLOS_RELAY_BOX"),
+ MastersPath: defaultMastersPath,
+ PressurePath: defaultPressurePath,
+ Store: store,
+ Limits: LimitsFromEnv(getenv),
+ }
+ if p := env("CARLOS_RELAY_MASTERS_PATH"); p != "" {
+ cfg.MastersPath = p
+ }
+ if p := env("CARLOS_RELAY_PRESSURE_PATH"); p != "" {
+ cfg.PressurePath = p
+ }
+
+ // The region rule is edgesFileRegion's, spelled again here rather than
+ // imported, because the two must agree: CARLOS_AWS_REGION wins outright,
+ // otherwise the IMDS placement probe stands. Empty is not fatal — such a
+ // relay matches no app's region list and serves STUN only, which is a
+ // legitimate state, not a misconfiguration to die on.
+ cfg.Region = env("CARLOS_AWS_REGION")
+ if cfg.Region == "" && imdsRegion != nil {
+ cfg.Region = strings.TrimSpace(imdsRegion())
+ }
+
+ ip := env("CARLOS_RELAY_PUBLIC_IP")
+ if ip == "" && imdsIP != nil {
+ ip = strings.TrimSpace(imdsIP())
+ }
+ if ip == "" {
+ return Config{}, errors.New("relay: no public IP: set CARLOS_RELAY_PUBLIC_IP, or run on a box whose IMDS reports a public-ipv4")
+ }
+ cfg.PublicIP = net.ParseIP(ip)
+ if cfg.PublicIP == nil {
+ return Config{}, fmt.Errorf("relay: CARLOS_RELAY_PUBLIC_IP %q is not an IP address", ip)
+ }
+
+ // A typo in the deny list must stop the relay, not quietly open the VPC
+ // to any browser that can reach 3478.
+ for _, s := range strings.Split(env("CARLOS_RELAY_DENY_CIDRS"), ",") {
+ s = strings.TrimSpace(s)
+ if s == "" {
+ continue
+ }
+ _, n, err := net.ParseCIDR(s)
+ if err != nil {
+ return Config{}, fmt.Errorf("relay: CARLOS_RELAY_DENY_CIDRS %q: %w", s, err)
+ }
+ cfg.DenyCIDRs = append(cfg.DenyCIDRs, n)
+ }
+ return cfg, nil
+}
+
+// Serve runs the relay until ctx is done, then closes every listener and
+// allocation and returns nil. The only errors it returns are from starting:
+// once it is listening, every failure is logged and retried on the next
+// tick, because a relay that exits on a bad bucket read is a relay that
+// stops answering STUN for a reason that has nothing to do with STUN.
+func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
+ if logf == nil {
+ logf = func(string, ...any) {}
+ }
+ switch {
+ case cfg.Box == "":
+ return errors.New("relay: Config.Box is required (the metrics box label)")
+ case cfg.Store == nil:
+ return errors.New("relay: Config.Store is required")
+ case cfg.PublicIP == nil:
+ return errors.New("relay: Config.PublicIP is required")
+ }
+ if cfg.MastersPath == "" {
+ cfg.MastersPath = defaultMastersPath
+ }
+ if cfg.Region == "" {
+ logf("relay: no region: this box matches no app's region list and will serve STUN only")
+ }
+
+ // The four live views the server reads on every request. They are
+ // written here, under mu, and read on pion's goroutines, so every
+ // accessor takes the lock and hands back a value the server may hold.
+ var (
+ mu sync.Mutex
+ masters *Masters
+ mastersE string
+ auth *Authority
+ lease Lease
+ )
+ now := time.Now
+ meter := NewMeter(cfg.Box, now, logf)
+
+ udp, tcp := cfg.udp, cfg.tcp
+ if udp == "" && tcp == "" {
+ udp, tcp = relayUDPAddr, relayTCPAddr
+ }
+ srv, err := NewServer(ServerConfig{
+ Region: cfg.Region,
+ RelayIP: cfg.PublicIP,
+ UDP: udp,
+ TCP: tcp,
+ MinPort: 49152,
+ MaxPort: 65535,
+ DenyCIDRs: cfg.DenyCIDRs,
+ Limits: cfg.Limits,
+ Now: now,
+ Log: logf,
+ Meter: meter,
+ Masters: func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
+ Authority: func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
+ Lease: func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
+ })
+ if err != nil {
+ return err
+ }
+ defer srv.Close()
+ logf("relay: listening on %v, relay address %s, region %q", srv.Listeners(), cfg.PublicIP, cfg.Region)
+
+ // Rebuild the rolling-hour byte windows from what this box already
+ // metered this hour and last (spec §3, "Limits"): a restart must not
+ // hand an app that has spent its 20 GB a fresh budget.
+ if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now(), logf); err != nil {
+ logf("relay: rolling-hour windows not rebuilt (%v); budgets start empty", err)
+ } else {
+ seedWindows(srv, samples)
+ }
+
+ go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) {
+ mu.Lock()
+ if err != nil {
+ mastersE = err.Error()
+ } else {
+ masters, mastersE = m, ""
+ }
+ mu.Unlock()
+ if err != nil {
+ logf("relay: masters file %s: %v", cfg.MastersPath, err)
+ return
+ }
+ // A reload can retire a master, and an allocation signed by it must
+ // end now rather than at the next expiry sweep.
+ srv.tick()
+ })
+
+ // readAuth is one authority read: it renews the lease on success only
+ // (a failed read renews nothing, spec §3), ends what the new record no
+ // longer allows, and writes the heartbeat that tells the console this
+ // box is current.
+ readAuth := func() {
+ a, _, err := ReadAuthority(ctx, cfg.Store)
+ t := now()
+ mu.Lock()
+ if err == nil {
+ auth = a
+ lease.Renew(t)
+ }
+ cur, live := auth, lease.Live(t)
+ hb := Heartbeat{
+ Box: cfg.Box,
+ Region: cfg.Region,
+ ReadAt: t,
+ Listeners: srv.Listeners(),
+ Kids: kidsOf(masters),
+ MastersError: mastersE,
+ }
+ if cur != nil {
+ hb.WrittenAt = cur.WrittenAt
+ }
+ mu.Unlock()
+ if err != nil {
+ logf("relay: authority read: %v", err)
+ }
+ if !live {
+ srv.EndAll("lease lapsed")
+ }
+ // cur is a snapshot, so the predicate runs off the lock without
+ // racing the next read. A nil record means no app is enabled here
+ // and every allocation goes, which is the "STUN only" state.
+ srv.EndWhere(func(b *Binding) bool {
+ x, ok := cur.Lookup(b.ID.Account, b.ID.App)
+ return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region)
+ }, "authority")
+ if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil {
+ logf("relay: heartbeat: %v", err)
+ }
+ }
+ readAuth()
+
+ flush := func(ctx context.Context) {
+ for id, n := range srv.LiveByIdentity() {
+ meter.SetAllocs(id, n)
+ }
+ if err := meter.Flush(ctx, cfg.Store); err != nil {
+ logf("relay: meter flush: %v", err)
+ }
+ }
+
+ authTick := time.NewTicker(authInterval)
+ serialTick := time.NewTicker(serialInterval)
+ meterTick := time.NewTicker(meterInterval)
+ expiryTick := time.NewTicker(expiryInterval)
+ psiTick := time.NewTicker(psiInterval)
+ defer func() {
+ for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} {
+ t.Stop()
+ }
+ }()
+
+ var (
+ lastSerial int64 = -1
+ shedding bool
+ )
+ for {
+ select {
+ case <-ctx.Done():
+ // The last minute of metering is real money; give it its own
+ // short deadline rather than losing it to the cancelled ctx.
+ final, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+ flush(final)
+ cancel()
+ return nil
+ case <-authTick.C:
+ readAuth()
+ case <-serialTick.C:
+ // The relay polls control/serial itself: the agent's
+ // fastConvergeLoop is a different process (spec §3).
+ doc, _, err := release.ReadSerial(ctx, cfg.Store)
+ switch {
+ case err != nil:
+ logf("relay: serial read: %v", err)
+ case doc.Serial != lastSerial:
+ lastSerial = doc.Serial
+ readAuth()
+ }
+ case <-meterTick.C:
+ flush(ctx)
+ case <-expiryTick.C:
+ srv.tick()
+ case <-psiTick.C:
+ some, err := readPSI(cfg.PressurePath)
+ if err != nil {
+ continue
+ }
+ // Spec §3 asks for shedding above 20% and a resumption of
+ // admission under 10%. The Server has no admission pause to
+ // lift — admission is governed by the limits, not by pressure —
+ // so "resume" here is simply the end of shedding: once over 20%
+ // the relay drops its oldest allocation every tick until
+ // pressure falls under 10%, and then stops.
+ switch {
+ case some > shedAbove:
+ shedding = true
+ case some < shedUntil:
+ shedding = false
+ }
+ if shedding {
+ srv.ShedOldest(1)
+ }
+ }
+ }
+}
+
+// seedWindows replays a window of metered samples into the admission
+// counters. A sample's app key is "<account>/<app>" and an app name can
+// never contain a slash (spec §2's grammar), so the first slash splits it.
+func seedWindows(srv *Server, samples []MeterSample) {
+ for _, s := range samples {
+ for key, app := range s.Apps {
+ account, name, ok := strings.Cut(key, "/")
+ if !ok || account == "" || name == "" {
+ continue
+ }
+ srv.Seed(Identity{Account: account, App: name}, s.At, app.In+app.Out)
+ }
+ }
+}
+
+// kidsOf is Masters.Kids with a nil guard: the heartbeat reports the kids
+// currently loaded, and "none, because the file has never parsed" is a
+// state the console wants to see rather than a panic.
+func kidsOf(m *Masters) []string {
+ if m == nil {
+ return nil
+ }
+ return m.Kids()
+}
+
+// readPSI reads the `some avg10` percentage out of a pressure-stall file
+// (/proc/pressure/memory):
+//
+// some avg10=0.00 avg60=0.00 avg300=0.00 total=0
+// full avg10=0.00 avg60=0.00 avg300=0.00 total=0
+//
+// Only the `some` line is used: it is the share of time ANY task stalled on
+// memory, which is what "the box is under pressure" means for a unit that
+// has to lose first. Plain file reading, so this compiles on the platforms
+// the CLI cross-builds for and simply errors there.
+func readPSI(path string) (float64, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return 0, err
+ }
+ defer f.Close()
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ fields := strings.Fields(sc.Text())
+ if len(fields) == 0 || fields[0] != "some" {
+ continue
+ }
+ for _, kv := range fields[1:] {
+ v, ok := strings.CutPrefix(kv, "avg10=")
+ if !ok {
+ continue
+ }
+ n, err := strconv.ParseFloat(v, 64)
+ if err != nil {
+ return 0, fmt.Errorf("%s: some avg10=%q: %w", path, v, err)
+ }
+ return n, nil
+ }
+ }
+ if err := sc.Err(); err != nil {
+ return 0, err
+ }
+ return 0, fmt.Errorf("%s: no `some avg10=` field", path)
+}
diff --git a/internal/relay/serve_test.go b/internal/relay/serve_test.go
new file mode 100644
index 000000000..95d7e9165
--- /dev/null
+++ b/internal/relay/serve_test.go
@@ -0,0 +1,276 @@
+package relay
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestReadPSI(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ body string
+ want float64
+ err bool
+ }{
+ {
+ name: "quiet box",
+ body: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n",
+ want: 0,
+ },
+ {
+ // The `full` line is deliberately higher: taking it instead of
+ // `some` would read as pressure the relay must not shed on.
+ name: "under pressure",
+ body: "some avg10=23.47 avg60=11.20 avg300=3.01 total=91234567\nfull avg10=44.00 avg60=9.00 avg300=1.00 total=44\n",
+ want: 23.47,
+ },
+ {name: "no some line", body: "full avg10=1.00 avg60=0 avg300=0 total=0\n", err: true},
+ {name: "unparseable", body: "some avg10=lots avg60=0 avg300=0 total=0\n", err: true},
+ {name: "empty", body: "", err: true},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "pressure")
+ if err := os.WriteFile(path, []byte(tc.body), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ got, err := readPSI(path)
+ if tc.err {
+ if err == nil {
+ t.Fatalf("readPSI(%q) = %v, want an error", tc.body, got)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("readPSI: %v", err)
+ }
+ if got != tc.want {
+ t.Fatalf("readPSI = %v, want %v", got, tc.want)
+ }
+ })
+ }
+}
+
+// A missing pressure file is the ordinary state on a kernel without PSI and
+// on the platforms this binary cross-compiles for: an error the caller skips
+// a tick on, never a panic.
+func TestReadPSIMissingFile(t *testing.T) {
+ if _, err := readPSI(filepath.Join(t.TempDir(), "nope")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("readPSI on a missing file = %v, want ErrNotExist", err)
+ }
+}
+
+func TestConfigFromEnvRefusesNoPublicIP(t *testing.T) {
+ _, err := ConfigFromEnv(context.Background(), func(string) string { return "" },
+ func() string { return "" }, func() string { return "eu-west-1" }, newMemStore(t))
+ if err == nil {
+ t.Fatal("ConfigFromEnv with no public IP anywhere: want an error")
+ }
+ if !strings.Contains(err.Error(), "CARLOS_RELAY_PUBLIC_IP") {
+ t.Fatalf("error should name the override: %v", err)
+ }
+}
+
+func TestConfigFromEnv(t *testing.T) {
+ env := map[string]string{
+ "CARLOS_RELAY_PUBLIC_IP": " 203.0.113.7 ",
+ "CARLOS_AWS_REGION": "us-east-1",
+ "CARLOS_RELAY_DENY_CIDRS": "10.4.0.0/16, 198.51.100.9/32",
+ "CARLOS_RELAY_MASTERS_PATH": "/tmp/masters",
+ "CARLOS_RELAY_BOX": "box-7",
+ "CARLOS_RELAY_LIMIT_APP_LIVE": "9",
+ "CARLOS_RELAY_PRESSURE_PATH": "/tmp/pressure",
+ "CARLOS_DEPLOYMENT_BUCKET_ETC": "ignored",
+ }
+ cfg, err := ConfigFromEnv(context.Background(), func(k string) string { return env[k] },
+ func() string { return "192.0.2.1" }, func() string { return "eu-west-1" }, newMemStore(t))
+ if err != nil {
+ t.Fatalf("ConfigFromEnv: %v", err)
+ }
+ // Both overrides must beat IMDS, and the region one must match
+ // edgesFileRegion's rule so the relay and the edges file agree.
+ if got := cfg.PublicIP.String(); got != "203.0.113.7" {
+ t.Errorf("PublicIP = %s, want the override", got)
+ }
+ if cfg.Region != "us-east-1" {
+ t.Errorf("Region = %s, want the override", cfg.Region)
+ }
+ if cfg.Box != "box-7" || cfg.MastersPath != "/tmp/masters" || cfg.PressurePath != "/tmp/pressure" {
+ t.Errorf("config = %+v", cfg)
+ }
+ if len(cfg.DenyCIDRs) != 2 || cfg.DenyCIDRs[0].String() != "10.4.0.0/16" || cfg.DenyCIDRs[1].String() != "198.51.100.9/32" {
+ t.Errorf("DenyCIDRs = %v", cfg.DenyCIDRs)
+ }
+ if cfg.Limits.AppLive != 9 {
+ t.Errorf("AppLive = %d, want the env limit", cfg.Limits.AppLive)
+ }
+}
+
+// A malformed deny list is an operator typo that would otherwise open the
+// VPC to anything that can reach 3478, so it stops the relay.
+func TestConfigFromEnvRefusesBadDenyCIDR(t *testing.T) {
+ env := map[string]string{"CARLOS_RELAY_PUBLIC_IP": "203.0.113.7", "CARLOS_RELAY_DENY_CIDRS": "10.4.0.0/16,not-a-cidr"}
+ if _, err := ConfigFromEnv(context.Background(), func(k string) string { return env[k] }, nil, nil, newMemStore(t)); err == nil {
+ t.Fatal("want an error for a malformed CIDR")
+ }
+}
+
+// With no override, IMDS supplies both, and an off-EC2 box with neither
+// region source is region-"" rather than an error: it serves STUN.
+func TestConfigFromEnvFallsBackToIMDS(t *testing.T) {
+ cfg, err := ConfigFromEnv(context.Background(), func(string) string { return "" },
+ func() string { return "192.0.2.5" }, func() string { return "" }, newMemStore(t))
+ if err != nil {
+ t.Fatalf("ConfigFromEnv: %v", err)
+ }
+ if cfg.PublicIP.String() != "192.0.2.5" || cfg.Region != "" || cfg.MastersPath != defaultMastersPath {
+ t.Fatalf("config = %+v", cfg)
+ }
+}
+
+func TestKidsOfNilMasters(t *testing.T) {
+ if got := kidsOf(nil); got != nil {
+ t.Fatalf("kidsOf(nil) = %v, want nil", got)
+ }
+ m, err := ParseMasters(strings.NewReader("k1 " + b64(testMaster()) + " -\nk2 " + b64(testMaster()) + " -\n"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := kidsOf(m); len(got) != 2 || got[0] != "k1" || got[1] != "k2" {
+ t.Fatalf("kidsOf = %v", got)
+ }
+}
+
+// seedWindows must charge a restart's rebuilt window with in+out, keyed on
+// the FIRST slash: an app name can never contain one, an account sqid never
+// can either, and a malformed key is skipped rather than seeding a garbage
+// identity.
+func TestSeedWindows(t *testing.T) {
+ srv, _, _, _, _ := newTestServer(t, "203.0.113.7", nil, func(cfg *ServerConfig) {
+ cfg.Limits = Limits{AppLive: 1, AppPerMinute: 1, AppBytesPerHour: 1000,
+ AccountLive: 1, AccountPerMinute: 1, AccountBytesPerHour: 1 << 40, BoxLive: 1, BoxPerMinute: 1}
+ })
+ at := time.Now().Add(-30 * time.Minute)
+ seedWindows(srv, []MeterSample{{
+ At: at,
+ Apps: map[string]MeterApp{
+ "abc/meet": {In: 400, Out: 300},
+ "abc/other": {In: 1, Out: 1},
+ "nokey": {In: 999999},
+ },
+ }})
+ meet := Identity{Account: "abc", App: "meet"}
+ if srv.adm.OverBudget(meet, time.Now()) {
+ t.Fatal("700 of 1000 bytes seeded should not be over budget")
+ }
+ seedWindows(srv, []MeterSample{{At: at, Apps: map[string]MeterApp{"abc/meet": {In: 400}}}})
+ if !srv.adm.OverBudget(meet, time.Now()) {
+ t.Fatal("1100 of 1000 bytes seeded should be over budget")
+ }
+ if srv.adm.OverBudget(Identity{Account: "nokey"}, time.Now()) {
+ t.Fatal("a key with no slash must not seed anything")
+ }
+}
+
+// The serve loop, end to end on ephemeral ports: it binds, reads the
+// authority record, writes the heartbeat the console reads, and shuts down
+// cleanly when its context is cancelled (systemd's SIGTERM).
+func TestServeWritesHeartbeatAndStops(t *testing.T) {
+ store := newMemStore(t)
+ dir := t.TempDir()
+ masters := filepath.Join(dir, "relay-masters")
+ if err := os.WriteFile(masters, []byte("k1 "+b64(testMaster())+" -\n"), 0o640); err != nil {
+ t.Fatal(err)
+ }
+ cfg := Config{
+ Box: "box-serve",
+ Region: "eu-west-1",
+ MastersPath: masters,
+ PublicIP: net.ParseIP("203.0.113.7"),
+ Store: store,
+ Limits: DefaultLimits(),
+ PressurePath: filepath.Join(dir, "no-pressure-here"),
+ udp: "127.0.0.1:0",
+ tcp: "127.0.0.1:0",
+ }
+ // The masters watcher outlives Serve by up to one poll interval, and
+ // t.Logf from a finished test is a data race, so the log goes quiet
+ // when the test body does.
+ var logMu sync.Mutex
+ quiet := false
+ logf := func(format string, args ...any) {
+ logMu.Lock()
+ defer logMu.Unlock()
+ if !quiet {
+ t.Logf(format, args...)
+ }
+ }
+ t.Cleanup(func() { logMu.Lock(); quiet = true; logMu.Unlock() })
+
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- Serve(ctx, cfg, logf) }()
+
+ hb := waitHeartbeat(t, store, "box-serve")
+ if hb.Region != "eu-west-1" || len(hb.Listeners) != 2 {
+ t.Errorf("heartbeat = %+v, want both listeners and the region", hb)
+ }
+ if hb.ReadAt.IsZero() {
+ t.Error("heartbeat carries no read time, so the lease clock is invisible")
+ }
+
+ cancel()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("Serve returned %v, want nil on a cancelled context", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("Serve did not return after its context was cancelled")
+ }
+}
+
+// Serve refuses to start without the two things it cannot invent: the box
+// label its objects are keyed by, and a store to write them to.
+func TestServeRefusesIncompleteConfig(t *testing.T) {
+ base := Config{Box: "b", Store: newMemStore(t), PublicIP: net.ParseIP("203.0.113.7")}
+ for name, mutate := range map[string]func(*Config){
+ "no box": func(c *Config) { c.Box = "" },
+ "no store": func(c *Config) { c.Store = nil },
+ "no ip": func(c *Config) { c.PublicIP = nil },
+ } {
+ cfg := base
+ mutate(&cfg)
+ if err := Serve(context.Background(), cfg, nil); err == nil {
+ t.Errorf("%s: Serve returned nil, want an error", name)
+ }
+ }
+}
+
+func waitHeartbeat(t *testing.T, store release.Store, box string) Heartbeat {
+ t.Helper()
+ deadline := time.Now().Add(10 * time.Second)
+ for {
+ body, _, err := store.Get(context.Background(), HeartbeatKey(box))
+ if err == nil {
+ var hb Heartbeat
+ if err := json.Unmarshal(body, &hb); err != nil {
+ t.Fatalf("heartbeat is not JSON: %v", err)
+ }
+ return hb
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("no heartbeat at %s within the deadline (last error %v)", HeartbeatKey(box), err)
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+}
diff --git a/internal/relay/server.go b/internal/relay/server.go
index 4e9bc4d48..671cc46fd 100644
--- a/internal/relay/server.go
+++ b/internal/relay/server.go
@@ -369,6 +369,21 @@ func (s *Server) Live() int {
return len(s.allocs)
}
+// LiveByIdentity is Live broken down per app, for the meter's `allocs`
+// gauge (spec §3, "Metering"). The Meter the server holds is narrowed to
+// Add, deliberately — the server counts bytes as it forwards them and knows
+// nothing about sample boundaries — so the gauge is the serve loop's job to
+// set, and this is where it reads it from.
+func (s *Server) LiveByIdentity() map[Identity]int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ out := make(map[Identity]int, len(s.allocs))
+ for _, e := range s.allocs {
+ out[e.id]++
+ }
+ return out
+}
+
// Seed replays metered bytes into the rolling-hour windows, so a restart
// resumes its budgets from the metrics objects it wrote.
func (s *Server) Seed(id Identity, at time.Time, n int64) { s.adm.Seed(id, at, n) }
diff --git a/main.go b/main.go
index fa80dc99c..f860f1dc5 100644
--- a/main.go
+++ b/main.go
@@ -75,6 +75,8 @@ func main() {
runEconomics(args)
case "store":
runStore(args)
+ case "relay":
+ runRelay(args)
case "email":
runEmail(args)
case "logs":
@@ -173,6 +175,8 @@ Your apps — ship and run software on a CARLOS deployment
--host or --environment narrows it, --dry-run names the set
schedule recurring or one-shot work per app: ls | set | rm | run
store object storage for an app: create | status | grant | rotate
+ relay the platform STUN/TURN relay: serve (what the unit runs)
+ and keygen (mint a region master)
email tenant email sending: enable | domains add | status | test |
credentials create/list/revoke | rotate | pause | resume
rollback point a channel back at an earlier version
diff --git a/relay.go b/relay.go
new file mode 100644
index 000000000..84191f8c6
--- /dev/null
+++ b/relay.go
@@ -0,0 +1,96 @@
+// relay.go is `carlos relay`: the platform STUN/TURN relay's own verbs
+// (relay spec §3-§5). Two of them exist so far and they sit at opposite
+// ends of the product:
+//
+// serve what carlos-relay.service runs. Not typed by hand.
+// keygen mint a region master. Typed by an operator, once per region
+// per rotation, and its output is a secret.
+//
+// The app-facing family (enable, status, disable, rotate, adopted) lands
+// beside them later; the usage line below grows with it.
+package main
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/base64"
+ "encoding/hex"
+ "fmt"
+ "log"
+ "os"
+ "os/signal"
+ "syscall"
+
+ "github.com/carlosframework/platform/internal/host"
+ "github.com/carlosframework/platform/internal/relay"
+)
+
+const relayUsage = `usage: carlos relay <serve|keygen> [flags]
+
+ serve run the relay (what carlos-relay.service execs; not typed by hand)
+ keygen mint a region master: one "<kid> <base64 key> -" line for
+ /etc/carlos/relay-masters and for the console's
+ CARLOS_RELAY_MASTERS_<REGION> secret`
+
+func runRelay(args []string) {
+ if len(args) == 0 {
+ fmt.Fprintln(os.Stderr, relayUsage)
+ os.Exit(2)
+ }
+ switch args[0] {
+ case "serve":
+ runRelayServe(args[1:])
+ case "keygen":
+ runRelayKeygen(args[1:])
+ case "-h", "--help", "help":
+ fmt.Println(relayUsage)
+ default:
+ fmt.Fprintln(os.Stderr, relayUsage)
+ os.Exit(2)
+ }
+}
+
+// runRelayServe is the unit's entry point. SIGTERM is systemd's stop, and
+// it must reach relay.Serve rather than killing the process, so the relay
+// closes its allocations and flushes its last minute of metering — the
+// bytes it forwarded in that minute are billable, and a hard kill loses
+// them.
+func runRelayServe(_ []string) {
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+ defer stop()
+
+ store, err := host.OpenDeployStore(ctx)
+ if err != nil {
+ log.Fatalf("relay serve: %v", err)
+ }
+ cfg, err := relay.ConfigFromEnv(ctx, os.Getenv, host.ReadPublicIP, host.ReadPlacementRegion, store)
+ if err != nil {
+ log.Fatalf("relay serve: %v", err)
+ }
+ // The box label is the metrics loop's, so the relay's heartbeat and its
+ // hourly metering objects land under the same name every other row this
+ // box publishes uses. It lives in internal/host, which internal/relay
+ // does not import, so it is filled in here.
+ if cfg.Box == "" {
+ cfg.Box = host.BoxLabel()
+ }
+ if err := relay.Serve(ctx, cfg, log.Printf); err != nil {
+ log.Fatalf("relay serve: %v", err)
+ }
+}
+
+// runRelayKeygen prints one masters-file line: a short key id, the key, and
+// "-" for "no retirement date". The key is printed once and never stored by
+// this process — the operator stages it on the region's boxes and puts the
+// same pair in the console's secret (spec §4).
+func runRelayKeygen(_ []string) {
+ key := make([]byte, relay.MasterKeyLen)
+ if _, err := rand.Read(key); err != nil {
+ log.Fatalf("relay keygen: %v", err)
+ }
+ kid := make([]byte, 4)
+ if _, err := rand.Read(kid); err != nil {
+ log.Fatalf("relay keygen: %v", err)
+ }
+ fmt.Printf("%s %s -\n", hex.EncodeToString(kid), base64.StdEncoding.EncodeToString(key))
+}