amadan

carlos / platform Public

edges: a box publishes its id, the HA capability, draining and headroom

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude c4ce3a8fa8275e72634a0e3e1de8c2cb13a0a8b7 parent fff400c
7 files changed, +363 −7
  • internal/console/instancesplacement.go +14 −4
  • internal/console/regioncandidates_test.go +27 −0
  • internal/host/edgesfile.go +79 −3
  • internal/host/edgesfile_test.go +152 −0
  • internal/host/headroom_linux.go +45 −0
  • internal/host/headroom_other.go +30 −0
  • internal/host/headroom_test.go +16 −0
diff --git a/internal/console/instancesplacement.go b/internal/console/instancesplacement.go
index 0b9510d44..7daa23ca3 100644
--- a/internal/console/instancesplacement.go
+++ b/internal/console/instancesplacement.go
@@ -59,10 +59,20 @@ type edgesFilePlacementView struct {
// predating them (unknown). SecretsKeyFP: absent = no key, or a writer
// predating it; AcceptsInbound's presence tells the two apart
// (nearest-region spec §6.1, R9).
- ServesSharedPool *bool `json:"serves_shared_pool"`
- AcceptsInbound *bool `json:"accepts_inbound"`
- SecretsKeyFP string `json:"secrets_key_fp"`
- Hosts map[string]json.RawMessage `json:"hosts"`
+ ServesSharedPool *bool `json:"serves_shared_pool"`
+ AcceptsInbound *bool `json:"accepts_inbound"`
+ SecretsKeyFP string `json:"secrets_key_fp"`
+ // Box evidence (region HA spec 1 §5.1). All absent on a writer that
+ // predates it: BoxID "", Capabilities nil, Draining false, Headroom nil.
+ BoxID string `json:"box_id"`
+ Capabilities []string `json:"capabilities"`
+ Draining bool `json:"draining"`
+ Headroom *struct {
+ MemFreeMiB int64 `json:"mem_free_mib"`
+ DiskFreeMiB int64 `json:"disk_free_mib"`
+ Awake int `json:"awake"`
+ } `json:"headroom"`
+ Hosts map[string]json.RawMessage `json:"hosts"`
// Fresh is not on the wire: readLiveEdgesSnapshot sets it when the
// box's own status snapshot is still fresh (status.BoxStale). An edges
// file is not liveness, see readLiveEdgesSnapshot.
diff --git a/internal/console/regioncandidates_test.go b/internal/console/regioncandidates_test.go
index 1687f38ce..afa0bdf6f 100644
--- a/internal/console/regioncandidates_test.go
+++ b/internal/console/regioncandidates_test.go
@@ -874,3 +874,30 @@ func TestRegionCandidatesCopyMatchesTheRecord(t *testing.T) {
t.Errorf("the record maps %s to regioncandidates.go, want O6,O7,O13,O14,O15", got)
}
}
+
+func TestEdgesViewParsesBoxEvidenceAndToleratesItsAbsence(t *testing.T) {
+ ctx, s := context.Background(), release.NewDirStore(t.TempDir())
+ with := `{"v":1,"box":"i-0abc","eip":"1.2.3.4","region":"eu-west-2","placement":"uk","hosts":{},"box_id":"i-0abc","capabilities":["region-ha-1"],"draining":true,"headroom":{"mem_free_mib":2210,"disk_free_mib":31200,"awake":14}}`
+ without := `{"v":1,"box":"old-box","eip":"5.6.7.8","region":"eu-west-1","hosts":{}}`
+ for k, b := range map[string]string{"i-0abc": with, "old-box": without} {
+ if _, err := s.Put(ctx, "control/edges/"+k+".json", []byte(b), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ snap, err := readEdgesSnapshot(ctx, s)
+ if err != nil || !snap.Complete {
+ t.Fatalf("snap complete=%v err=%v", snap.Complete, err)
+ }
+ by := map[string]edgesFilePlacementView{}
+ for _, f := range snap.Files {
+ by[f.Box] = f
+ }
+ n := by["i-0abc"]
+ if n.BoxID != "i-0abc" || len(n.Capabilities) != 1 || !n.Draining || n.Headroom == nil || n.Headroom.Awake != 14 {
+ t.Fatalf("new box parsed as %+v", n)
+ }
+ o := by["old-box"]
+ if o.BoxID != "" || o.Capabilities != nil || o.Draining || o.Headroom != nil {
+ t.Fatalf("old box parsed as %+v", o)
+ }
+}
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index 28fd105f1..f25501684 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -15,6 +15,7 @@ import (
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
"github.com/carlosframework/platform/internal/domains"
)
@@ -138,6 +139,51 @@ func servedOwners(routes []registry.Route, served map[string]edgeHost, paths []r
return owners
}
+// edgeHeadroom is this box's spare capacity at the time the file was
+// written. Zero values mean "could not read", never "none left": a chooser
+// ranks an unreadable box last, it does not exclude it.
+type edgeHeadroom struct {
+ MemFreeMiB int64 `json:"mem_free_mib"`
+ DiskFreeMiB int64 `json:"disk_free_mib"`
+ Awake int `json:"awake"`
+}
+
+const capRegionHA1 = "region-ha-1"
+
+// edgesFileBoxID is the instance id on EC2 and the label everywhere else.
+func edgesFileBoxID(instanceID, label string) string {
+ if instanceID != "" {
+ return instanceID
+ }
+ return label
+}
+
+// edgesFileCapabilities publishes region-ha-1 only when CARLOS_REGION_HA is
+// set to something other than "" or "0".
+func edgesFileCapabilities(getenv func(string) string) []string {
+ switch strings.TrimSpace(getenv("CARLOS_REGION_HA")) {
+ case "", "0":
+ return nil
+ }
+ return []string{capRegionHA1}
+}
+
+// awakeCount is how many routes this box has running now. An error reads
+// as zero: headroom is advisory.
+func awakeCount(reg *registry.Registry) int {
+ routes, err := reg.List()
+ if err != nil {
+ return 0
+ }
+ n := 0
+ for _, r := range routes {
+ if r.Status == "ready" && !r.Hibernate {
+ n++
+ }
+ }
+ return n
+}
+
// edgesFile is the wire shape at edgesFileKey(box), schema v1 (spec Part 3):
//
// {"v":1,"box":"quiet-harbour","eip":"203.0.113.10","region":"eu-west-1","generation":"2026-08-22T09:00:00Z","hosts":{"app.example.com":{"role":"home"}}}
@@ -226,6 +272,20 @@ type edgesFile struct {
// R9). The console requires it to equal the fingerprint of its own
// CARLOS_SECRETS_PUBKEY before it places anything here automatically.
SecretsKeyFP string `json:"secrets_key_fp,omitempty"`
+ // BoxID is the name claims use for this box (region HA spec 1 §2): the
+ // EC2 instance id, or the label off EC2. It is NOT evidence of identity:
+ // that is the box record (internal/boxid). A reader compares the two and
+ // treats a mismatch as an ineligible box.
+ BoxID string `json:"box_id,omitempty"`
+ // Capabilities gates multi-box regions (spec §8): "region-ha-1" once
+ // CARLOS_REGION_HA is on.
+ Capabilities []string `json:"capabilities,omitempty"`
+ // Draining is set while the placer empties this box (spec 4).
+ Draining bool `json:"draining,omitempty"`
+ // Headroom is what placement chooses by (spec 2). It is refreshed when
+ // the file is written for another reason and on the loop's own tick; a
+ // headroom change alone never forces a write (see edgesFilePass).
+ Headroom *edgeHeadroom `json:"headroom,omitempty"`
Generation string `json:"generation"`
Hosts map[string]edgeHost `json:"hosts"`
@@ -271,6 +331,15 @@ var withheldAliasSaid sync.Map
// once by the loop; eip is passed per pass because it is re-probed every
// tick and held across blanks (below). Hosts are derived here.
func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte) error {
+ return edgesFileWrite(ctx, s, reg, ident, eip, last, false)
+}
+
+// edgesFileWrite is edgesFilePass with one more input. refresh is true on
+// the loop's own timed pass: the file is then written even if nothing but
+// its headroom reading changed, so headroom is never more than one timer
+// period stale. On every other pass a headroom change alone is not a
+// reason to write.
+func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool) error {
box := ident.Box
routes, err := reg.List()
if err != nil {
@@ -484,11 +553,13 @@ func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry,
f := ident
f.V, f.EIP, f.Hosts, f.Generation = 1, eip, hosts, ""
- gate, err := json.Marshal(f)
+ gateCopy := f
+ gateCopy.Headroom = nil // a changing reading is not, by itself, a reason to write
+ gate, err := json.Marshal(gateCopy)
if err != nil {
return fmt.Errorf("edges: marshal: %w", err)
}
- if bytes.Equal(gate, *last) {
+ if bytes.Equal(gate, *last) && !(refresh && f.Headroom != nil) {
return nil
}
@@ -536,6 +607,7 @@ func seedEdgesFileGate(ctx context.Context, s release.Store, box string) []byte
return nil
}
prior.Generation = ""
+ prior.Headroom = nil // the gate never includes headroom: see edgesFileWrite
gate, err := json.Marshal(prior)
if err != nil {
return nil
@@ -661,7 +733,10 @@ func edgesFileLoop(reg *registry.Registry) {
ServesSharedPool: sharedPool,
AcceptsInbound: inbound,
SecretsKeyFP: edgesFileSecretsFP(os.Getenv),
+ BoxID: edgesFileBoxID(status.InstanceID(), box),
+ Capabilities: edgesFileCapabilities(os.Getenv),
}
+ dataDir := provisionTickConfig(os.Getenv).DataDir
log.Printf("agent: edges file loop armed (5min + jitter, and %s after a registry change, %s)", edgesFileWatchInterval, edgesFileKey(box))
last := seedEdgesFileGate(ctx, store, box)
sched := newEdgesFileSchedule(time.Now(), func() time.Duration {
@@ -692,7 +767,8 @@ func edgesFileLoop(reg *registry.Registry) {
if sched.timed {
eip = readPublicIP()
}
- err = edgesFilePass(ctx, store, reg, ident, eip, &last)
+ ident.Headroom = readHeadroom(dataDir, awakeCount(reg))
+ err = edgesFileWrite(ctx, store, reg, ident, eip, &last, sched.timed)
if err != nil {
// Throttled: a bucket that refuses every write (an IAM Deny, an
// outage) is retried on a backoff, and each retry must not be a
diff --git a/internal/host/edgesfile_test.go b/internal/host/edgesfile_test.go
index 7e8e03267..50051d133 100644
--- a/internal/host/edgesfile_test.go
+++ b/internal/host/edgesfile_test.go
@@ -1326,3 +1326,155 @@ func TestEdgesFileSecretsKeyFingerprint(t *testing.T) {
}
}
}
+
+func TestEdgesFileBoxIDIsTheInstanceIDWhenThereIsOne(t *testing.T) {
+ if got := edgesFileBoxID("i-0abc", "annie-welding"); got != "i-0abc" {
+ t.Fatalf("on EC2: %q", got)
+ }
+ if got := edgesFileBoxID("", "annie-welding"); got != "annie-welding" {
+ t.Fatalf("off EC2: %q", got)
+ }
+}
+
+func TestEdgesFileCapabilityFollowsTheFlag(t *testing.T) {
+ env := func(v string) func(string) string {
+ return func(k string) string {
+ if k == "CARLOS_REGION_HA" {
+ return v
+ }
+ return ""
+ }
+ }
+ if got := edgesFileCapabilities(env("")); len(got) != 0 {
+ t.Fatalf("unset: %v", got)
+ }
+ if got := edgesFileCapabilities(env("1")); len(got) != 1 || got[0] != "region-ha-1" {
+ t.Fatalf("set: %v", got)
+ }
+ if got := edgesFileCapabilities(env("0")); len(got) != 0 {
+ t.Fatalf("\"0\" is off: %v", got)
+ }
+}
+
+func TestEdgesFileRawWireBodyCarriesBoxEvidence(t *testing.T) {
+ s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t)
+ seedRoute(t, reg, "app.example.com")
+ var last []byte
+ ident := edgesFile{Box: "annie", BoxID: "i-0abc", Capabilities: []string{"region-ha-1"},
+ Headroom: &edgeHeadroom{MemFreeMiB: 2210, DiskFreeMiB: 31200, Awake: 14}}
+ if err := edgesFilePass(context.Background(), s, reg, ident, "1.2.3.4", &last); err != nil {
+ t.Fatal(err)
+ }
+ body, _, err := s.Get(context.Background(), "control/edges/annie.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var raw map[string]json.RawMessage
+ if err := json.Unmarshal(body, &raw); err != nil {
+ t.Fatal(err)
+ }
+ for key, want := range map[string]string{
+ "box_id": `"i-0abc"`,
+ "capabilities": `["region-ha-1"]`,
+ "headroom": `{"mem_free_mib":2210,"disk_free_mib":31200,"awake":14}`,
+ } {
+ if string(raw[key]) != want {
+ t.Errorf("%s = %s, want %s", key, raw[key], want)
+ }
+ }
+ if _, ok := raw["draining"]; ok {
+ t.Error("draining:false must be omitted")
+ }
+}
+
+func TestEdgesFileOmitsBoxEvidenceWhenAbsent(t *testing.T) {
+ s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t)
+ var last []byte
+ if err := edgesFilePass(context.Background(), s, reg, edgesFile{Box: "annie"}, "1.2.3.4", &last); err != nil {
+ t.Fatal(err)
+ }
+ body, _, _ := s.Get(context.Background(), "control/edges/annie.json")
+ for _, key := range []string{"box_id", "capabilities", "draining", "headroom"} {
+ if bytes.Contains(body, []byte(`"`+key+`"`)) {
+ t.Errorf("%s present in %s", key, body)
+ }
+ }
+}
+
+func TestEdgesFileHeadroomAloneDoesNotRewriteTheFile(t *testing.T) {
+ st := &putCountingStore{Store: release.NewDirStore(t.TempDir())}
+ reg := edgesTestRegistry(t)
+ seedRoute(t, reg, "app.example.com")
+ var last []byte
+ ident := edgesFile{Box: "annie", Headroom: &edgeHeadroom{MemFreeMiB: 100}}
+ if err := edgesFilePass(context.Background(), st, reg, ident, "1.2.3.4", &last); err != nil {
+ t.Fatal(err)
+ }
+ ident.Headroom = &edgeHeadroom{MemFreeMiB: 99}
+ if err := edgesFilePass(context.Background(), st, reg, ident, "1.2.3.4", &last); err != nil {
+ t.Fatal(err)
+ }
+ if st.puts != 1 {
+ t.Fatalf("puts = %d, want 1: a headroom change alone is not a reason to write", st.puts)
+ }
+}
+
+// The loop's timed pass refreshes headroom even when nothing else moved.
+func TestEdgesFileTimedPassRefreshesHeadroom(t *testing.T) {
+ st := &putCountingStore{Store: release.NewDirStore(t.TempDir())}
+ reg := edgesTestRegistry(t)
+ seedRoute(t, reg, "app.example.com")
+ var last []byte
+ ident := edgesFile{Box: "annie", Headroom: &edgeHeadroom{MemFreeMiB: 100}}
+ if err := edgesFileWrite(context.Background(), st, reg, ident, "1.2.3.4", &last, false); err != nil {
+ t.Fatal(err)
+ }
+ ident.Headroom = &edgeHeadroom{MemFreeMiB: 42}
+ if err := edgesFileWrite(context.Background(), st, reg, ident, "1.2.3.4", &last, true); err != nil {
+ t.Fatal(err)
+ }
+ if st.puts != 2 {
+ t.Fatalf("puts = %d, want 2: the timed pass must publish the new reading", st.puts)
+ }
+ body, _, err := st.Get(context.Background(), "control/edges/annie.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var f edgesFile
+ if err := json.Unmarshal(body, &f); err != nil || f.Headroom == nil || f.Headroom.MemFreeMiB != 42 {
+ t.Fatalf("published headroom = %+v, err %v", f.Headroom, err)
+ }
+ // A box that publishes no headroom keeps today's behaviour exactly:
+ // a timed pass with nothing changed writes nothing.
+ plain := &putCountingStore{Store: release.NewDirStore(t.TempDir())}
+ var plainLast []byte
+ for i := 0; i < 2; i++ {
+ if err := edgesFileWrite(context.Background(), plain, reg, edgesFile{Box: "betty"}, "1.2.3.4", &plainLast, true); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if plain.puts != 1 {
+ t.Fatalf("puts = %d, want 1 for a box with no headroom", plain.puts)
+ }
+}
+
+// A restart must not rewrite an unchanged file just because the stored
+// copy carries a headroom reading.
+func TestEdgesFileSeededGateIgnoresStoredHeadroom(t *testing.T) {
+ st := &putCountingStore{Store: release.NewDirStore(t.TempDir())}
+ reg := edgesTestRegistry(t)
+ seedRoute(t, reg, "app.example.com")
+ var last []byte
+ ident := edgesFile{Box: "annie", Headroom: &edgeHeadroom{MemFreeMiB: 100}}
+ if err := edgesFilePass(context.Background(), st, reg, ident, "1.2.3.4", &last); err != nil {
+ t.Fatal(err)
+ }
+ seeded := seedEdgesFileGate(context.Background(), st, "annie")
+ ident.Headroom = &edgeHeadroom{MemFreeMiB: 7}
+ if err := edgesFilePass(context.Background(), st, reg, ident, "1.2.3.4", &seeded); err != nil {
+ t.Fatal(err)
+ }
+ if st.puts != 1 {
+ t.Fatalf("puts = %d, want 1", st.puts)
+ }
+}
diff --git a/internal/host/headroom_linux.go b/internal/host/headroom_linux.go
new file mode 100644
index 000000000..6a23ca9c0
--- /dev/null
+++ b/internal/host/headroom_linux.go
@@ -0,0 +1,45 @@
+//go:build linux
+
+package host
+
+import (
+ "os"
+ "strconv"
+ "strings"
+
+ "golang.org/x/sys/unix"
+)
+
+// readHeadroom reads spare memory from /proc/meminfo and spare disk from
+// the data directory's filesystem. A reading that fails is left at zero.
+func readHeadroom(dataDir string, awake int) *edgeHeadroom {
+ h := &edgeHeadroom{Awake: awake}
+ if b, err := os.ReadFile("/proc/meminfo"); err == nil {
+ h.MemFreeMiB = parseMemAvailableMiB(string(b))
+ }
+ var st unix.Statfs_t
+ if err := unix.Statfs(dataDir, &st); err == nil {
+ h.DiskFreeMiB = int64(st.Bavail) * int64(st.Bsize) / (1 << 20)
+ }
+ return h
+}
+
+// parseMemAvailableMiB returns the MemAvailable line in MiB, or 0.
+func parseMemAvailableMiB(meminfo string) int64 {
+ for _, line := range strings.Split(meminfo, "\n") {
+ rest, ok := strings.CutPrefix(line, "MemAvailable:")
+ if !ok {
+ continue
+ }
+ fields := strings.Fields(rest)
+ if len(fields) == 0 {
+ return 0
+ }
+ kb, err := strconv.ParseInt(fields[0], 10, 64)
+ if err != nil {
+ return 0
+ }
+ return kb / 1024
+ }
+ return 0
+}
diff --git a/internal/host/headroom_other.go b/internal/host/headroom_other.go
new file mode 100644
index 000000000..84d4fa384
--- /dev/null
+++ b/internal/host/headroom_other.go
@@ -0,0 +1,30 @@
+//go:build !linux
+
+package host
+
+import (
+ "strconv"
+ "strings"
+)
+
+// readHeadroom has no readings off Linux: only the awake count.
+func readHeadroom(dataDir string, awake int) *edgeHeadroom { return &edgeHeadroom{Awake: awake} }
+
+func parseMemAvailableMiB(meminfo string) int64 {
+ for _, line := range strings.Split(meminfo, "\n") {
+ rest, ok := strings.CutPrefix(line, "MemAvailable:")
+ if !ok {
+ continue
+ }
+ fields := strings.Fields(rest)
+ if len(fields) == 0 {
+ return 0
+ }
+ kb, err := strconv.ParseInt(fields[0], 10, 64)
+ if err != nil {
+ return 0
+ }
+ return kb / 1024
+ }
+ return 0
+}
diff --git a/internal/host/headroom_test.go b/internal/host/headroom_test.go
new file mode 100644
index 000000000..c5c870160
--- /dev/null
+++ b/internal/host/headroom_test.go
@@ -0,0 +1,16 @@
+package host
+
+import "testing"
+
+func TestParseMemAvailable(t *testing.T) {
+ const meminfo = "MemTotal: 16000000 kB\nMemFree: 500000 kB\nMemAvailable: 2263040 kB\n"
+ if got := parseMemAvailableMiB(meminfo); got != 2210 {
+ t.Fatalf("got %d, want 2210", got)
+ }
+ if got := parseMemAvailableMiB("MemTotal: 1 kB\n"); got != 0 {
+ t.Fatalf("no MemAvailable line: got %d, want 0", got)
+ }
+ if got := parseMemAvailableMiB("MemAvailable: lots kB\n"); got != 0 {
+ t.Fatalf("unparseable: got %d, want 0", got)
+ }
+}