amadan

carlos / platform Public

status: per-upstream health in the box snapshot

The edge's probe table now reaches the box's own snapshot object:
Metrics.Upstreams, route host -> each upstream's state, omitempty so a box
with no probed route (nearly all of them) writes the identical bytes it
wrote before and every existing snapshot consumer parses unchanged.

The wire type lives in internal/status, not internal/edge. Its readers are
the console and operators — internal/console already parses BoxMetrics —
and none of them may be made to import the edge's ACME/DNS/proxy stack just
to name the shape of a struct they decode. internal/edge aliases it
(type UpstreamState = status.UpstreamState), so the prober keeps writing
UpstreamState in its own package with no conversion at the boundary, and
the dependency runs edge -> status, leaving internal/status a leaf over
registry/release.

Wiring follows tunnel state exactly: statusLoop starts before the edge
exists, so RunAgent's edge-ready callback publishes e.UpstreamStates
through an atomic source that the heartbeat reads each tick. The table is
taken WHOLE every tick and never merged with the last one, which is the
staleness contract — a route the box has stopped probing is simply absent
from the next snapshot rather than a ghost someone must remember to evict.

Privacy: Upstreams is route_versions' sibling and then some (hostname keys,
backend addresses in the values), so it goes in the same private
Box.Metrics block, guarded the same two ways — an allowlist entry that
pins the per-upstream fields, and a DryRun needle pair required inside
status/boxes/*.json and banned everywhere else the page publishes.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 6c0053e9fca7a5c3f9c3e3e3317cc6f67fab80f3 parent 9a3c525
9 files changed, +475 −25
  • internal/edge/edge.go +17 −0
  • internal/edge/edge_test.go +70 −0
  • internal/edge/prober.go +10 −9
  • internal/host/agent.go +53 −0
  • internal/host/host_test.go +92 −0
  • internal/status/contract_test.go +42 −9
  • internal/status/dryrun.go +45 −0
  • internal/status/snapshot.go +67 −7
  • internal/status/status_test.go +79 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index c7b0768c1..1a981ea9a 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -611,6 +611,23 @@ func (e *Edge) currentHosts() []string {
return out
}
+// UpstreamStates is the probe table as the rest of the process may see it:
+// route host -> each upstream's current state, in the route's addr order,
+// for probed (multi-upstream) routes only. A route with no health config
+// holds no prober and is absent, as is every host on an edge that probes
+// nothing at all — the common shape, which reports an empty map.
+//
+// The agent's status heartbeat is the caller (internal/host wires this in from
+// the edge-ready callback, the same seam tunnelLoop arrives through) and it
+// calls this once per tick, taking the WHOLE table each time rather than
+// diffing: everything returned is a copy proberSet.State built under the
+// per-host lock, so serializing it while probes keep running is safe, and a
+// route that stopped being probed is simply gone from the next call rather
+// than lingering as a stale entry someone has to remember to evict.
+func (e *Edge) UpstreamStates() map[string][]UpstreamState {
+ return e.probers.State()
+}
+
// Allowed reports whether host is configured — used as the autocert
// HostPolicy, so a certificate is only ever requested for a host we serve.
func (e *Edge) Allowed(host string) bool {
diff --git a/internal/edge/edge_test.go b/internal/edge/edge_test.go
index 76ae05b87..e5102eaef 100644
--- a/internal/edge/edge_test.go
+++ b/internal/edge/edge_test.go
@@ -1755,6 +1755,76 @@ func TestSingleAddrWithHealthGets503Semantics(t *testing.T) {
}
}
+// TestUpstreamStatesExportsProbeTable pins Task 6's accessor: the box status
+// snapshot's only door onto the probe table is Edge.UpstreamStates, which must
+// report exactly what the unexported proberSet holds — same hosts, same addrs,
+// same order — and must be safe to call on an edge with nothing probed (a
+// pure-proxy `carlos edge`, which is most of them), where it reports empty
+// rather than nil-panicking or inventing a host.
+func TestUpstreamStatesExportsProbeTable(t *testing.T) {
+ addrA, _ := countingBackend(t, "A")
+ dead := deadAddr(t)
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "ha.example.com", Kind: "service", Status: "ready",
+ Addrs: addrA + "," + dead,
+ Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ // A second, UNPROBED route: no health config, so it holds no prober and
+ // must not appear in the exported table at all.
+ if err := reg.Upsert(registry.Route{
+ Host: "plain.example.com", Kind: "service", Status: "ready", Addr: addrA,
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+
+ got := e.UpstreamStates()
+ if len(got) != 1 {
+ t.Fatalf("UpstreamStates() = %+v, want exactly the one probed host", got)
+ }
+ states := got["ha.example.com"]
+ if len(states) != 2 || states[0].Addr != addrA || states[1].Addr != dead {
+ t.Fatalf("states = %+v, want [%s %s] in the route's addr order", states, addrA, dead)
+ }
+ // Optimistic start (prober.go rule 1) is what the snapshot reports for a
+ // freshly synced host — frozenHealth cannot have ticked a transition in.
+ for _, st := range states {
+ if !st.Up || st.Since.IsZero() {
+ t.Errorf("state %+v: a freshly synced upstream must be up with a Since stamp", st)
+ }
+ }
+
+ // The exported view is a COPY: mutating it must not reach the live table
+ // the request path picks from.
+ states[0].Up = false
+ if again := e.UpstreamStates()["ha.example.com"]; !again[0].Up {
+ t.Error("UpstreamStates() handed out a view of live state — the snapshot must never be able to eject an upstream")
+ }
+
+ // An edge with nothing probed answers empty, not nil-panic.
+ plainReg := openReg(t)
+ if err := plainReg.Upsert(registry.Route{Host: "plain.example.com", Kind: "service", Status: "ready", Addr: addrA}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ plain, err := New(plainReg, nil)
+ if err != nil {
+ t.Fatalf("New (unprobed): %v", err)
+ }
+ defer plain.probers.Stop()
+ if st := plain.UpstreamStates(); len(st) != 0 {
+ t.Errorf("UpstreamStates() on an unprobed edge = %+v, want empty", st)
+ }
+}
+
func TestLegacySingleAddrRouteUnchanged(t *testing.T) {
dead := deadAddr(t)
diff --git a/internal/edge/prober.go b/internal/edge/prober.go
index 9467c0d58..b6b0b6de8 100644
--- a/internal/edge/prober.go
+++ b/internal/edge/prober.go
@@ -40,6 +40,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/status"
)
// probeTimeout bounds a single probe request end to end (connect through
@@ -72,15 +73,15 @@ const (
)
// UpstreamState is the dashboard-facing view of one upstream: what the probe
-// currently believes, when that belief last changed, and why. Task 6 carries
-// this into the box status snapshot, so the JSON tags are wire shape.
-type UpstreamState struct {
- Addr string `json:"addr"`
- Up bool `json:"up"`
- Since time.Time `json:"since"` // last transition (or the sync that started this prober)
- Fails int `json:"fails"` // consecutive probe failures
- LastError string `json:"last_error,omitempty"` // dial vs status, for the log/dashboard
-}
+// currently believes, when that belief last changed, and why.
+//
+// An ALIAS, not a declaration: the struct itself lives in internal/status
+// (Task 6), because it is a wire type on status/boxes/<label>.json and its
+// readers — the console, operators — must be able to decode it without
+// importing the edge. See status.UpstreamState's own doc comment for the full
+// argument. The alias keeps this package writing plain UpstreamState with no
+// conversion where the prober fills it in.
+type UpstreamState = status.UpstreamState
// probeSpec is one route's probing contract, parsed from the registry's health
// JSON once per Reload. Health need not have been through WithDefaults: Sync
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 23fb23f62..5735d2190 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -136,6 +136,7 @@ import (
"sort"
"strings"
"sync"
+ "sync/atomic"
"time"
"github.com/carlosframework/platform/internal/activator"
@@ -315,6 +316,12 @@ func RunAgent(args []string) {
// calling it directly would block the front door forever.
edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
go tunnelLoop(e)
+ // The edge owns the health probers, so its state table is only
+ // reachable once it exists — statusLoop (already running above) reads
+ // it through this published source on its next tick. Method value, so
+ // the loop always sees the CURRENT probe set: Reload replaces probers
+ // behind this handle without the heartbeat needing to be re-wired.
+ setUpstreamSource(e.UpstreamStates)
})
// The edge has drained; take the children down too (they're this process's
@@ -1601,6 +1608,46 @@ func shouldSkipPublisherElection(getenv func(string) string) bool {
return getenv("CARLOS_FLEET") != ""
}
+// upstreamSource holds the live edge's UpstreamStates method, published by
+// RunAgent's edge-ready callback and read by statusLoop's probe set — the same
+// shape tunnelStateVal uses for tunnel state (tunnel_client.go), and for the
+// same reason: the status loop starts BEFORE the edge exists (RunAgent spawns
+// the loops, then builds and serves the edge), so the two cannot be wired by
+// passing a handle down.
+//
+// A func rather than the *edge.Edge itself, so what this package's tests
+// inject is the same one value production sets — no fake Edge, no probe
+// goroutine a test would have no way to stop.
+var upstreamSource atomic.Value // func() map[string][]status.UpstreamState
+
+// setUpstreamSource publishes (or, with nil, withdraws) the probe-state
+// source. Called once, from the edge-ready callback.
+func setUpstreamSource(fn func() map[string][]status.UpstreamState) {
+ if fn == nil {
+ upstreamSource.Store(func() map[string][]status.UpstreamState { return nil })
+ return
+ }
+ upstreamSource.Store(fn)
+}
+
+// upstreamStates reads the current probe table for the heartbeat, or nil when
+// no source has been published yet.
+//
+// nil is the honest answer for every box shape that has no edge probe table to
+// report: `carlos edge` run without the agent, an agent whose ready callback
+// hasn't fired yet, and the overwhelmingly common case of a box with no
+// multi-upstream route at all. All three publish no upstreams key rather than
+// an empty object (status.CollectBox's omitempty rule) — a box that isn't
+// probing and a box running code that predates this field stay
+// indistinguishable on the wire.
+func upstreamStates() map[string][]status.UpstreamState {
+ fn, _ := upstreamSource.Load().(func() map[string][]status.UpstreamState)
+ if fn == nil {
+ return nil
+ }
+ return fn()
+}
+
// statusLoop runs the fleet status page's 5-minute tick, forever, once armed
// — see the package doc comment above for the full env contract. Missing
// CARLOS_STATUS_BUCKET disables the loop with one log line, the same
@@ -1766,6 +1813,12 @@ func statusLoop(reg *registry.Registry) {
DiskFreeBytes: readDiskFreeBytes(),
Release: version.Build,
Tunnel: tunnelState(),
+ // The edge's probe table, read WHOLE every tick (never merged
+ // with the last one), so a route this box has stopped probing
+ // leaves no ghost behind in the next snapshot. nil until the
+ // edge-ready callback publishes the source, and on any box that
+ // probes nothing — both publish no key at all.
+ Upstreams: upstreamStates(),
}
// A fleet box (CARLOS_FLEET set) never gets here as a contender: it
diff --git a/internal/host/host_test.go b/internal/host/host_test.go
index bfe8cdec7..09cd6a18e 100644
--- a/internal/host/host_test.go
+++ b/internal/host/host_test.go
@@ -462,6 +462,98 @@ func TestFleetBoxHeartbeatLandsUnderFleetPrefixNeverPublishes(t *testing.T) {
}
}
+// ---- Task 6 (multi-upstream plan): per-upstream probe state in the snapshot -
+
+// TestStatusTickCarriesProbedUpstreams is Task 6's Step-1 test: the state the
+// edge's probers hold for a multi-upstream route must reach the box's OWN
+// snapshot object through the real tick — statusTick over a DirStore, read
+// back as the bytes an operator (or the console) would parse — not a
+// hand-built Box.
+//
+// It drives the same seam production does: setUpstreamSource is what RunAgent
+// hands the live *edge.Edge's UpstreamStates method to from the edge-ready
+// callback, and upstreamStates() is what statusLoop reads on every tick. The
+// fixture stands in for the Edge here for the same reason every other probe in
+// this file is injected — a real prober would need a live listener and a
+// goroutine this package cannot stop.
+func TestStatusTickCarriesProbedUpstreams(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "ha.example.com", Kind: "service", Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+
+ store := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ since := time.Now().Add(-90 * time.Second).UTC().Truncate(time.Second)
+
+ setUpstreamSource(func() map[string][]status.UpstreamState {
+ return map[string][]status.UpstreamState{"ha.example.com": {
+ {Addr: "10.0.0.11:8080", Up: true, Since: since},
+ {Addr: "10.0.0.12:8080", Up: false, Since: since, Fails: 3, LastError: "dial: connection refused"},
+ }}
+ })
+ t.Cleanup(func() { setUpstreamSource(nil) })
+
+ // Exactly what statusLoop builds each tick.
+ probes := status.Probes{MachineID: "test-machine", Upstreams: upstreamStates()}
+ if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
+ t.Fatalf("statusTick: %v", err)
+ }
+
+ key := "status/boxes/" + status.Label("test-machine") + ".json"
+ raw, _, err := store.Get(ctx, key)
+ if err != nil {
+ t.Fatalf("Get %s: %v", key, err)
+ }
+ var box status.Box
+ if err := json.Unmarshal(raw, &box); err != nil {
+ t.Fatalf("unmarshal %s: %v", key, err)
+ }
+ if box.Metrics == nil {
+ t.Fatalf("metrics block missing from %s: %s", key, raw)
+ }
+ got := box.Metrics.Upstreams["ha.example.com"]
+ want := []status.UpstreamState{
+ {Addr: "10.0.0.11:8080", Up: true, Since: since},
+ {Addr: "10.0.0.12:8080", Up: false, Since: since, Fails: 3, LastError: "dial: connection refused"},
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Errorf("upstreams for ha.example.com = %+v, want %+v: %s", got, want, raw)
+ }
+
+ // Staleness: the map is rebuilt from the prober's State() every tick, so a
+ // box that stops probing a route (route deleted, addrs cleared) must leave
+ // NO ghost entry behind in the next snapshot — and with nothing probed at
+ // all the key must be absent, not an empty object.
+ setUpstreamSource(func() map[string][]status.UpstreamState { return nil })
+ probes.Upstreams = upstreamStates()
+ if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
+ t.Fatalf("statusTick (no probers): %v", err)
+ }
+ raw, _, err = store.Get(ctx, key)
+ if err != nil {
+ t.Fatalf("Get %s after second tick: %v", key, err)
+ }
+ if strings.Contains(string(raw), "ha.example.com") || strings.Contains(string(raw), "upstreams") {
+ t.Errorf("second snapshot still carries a retired route's probe state: %s", raw)
+ }
+}
+
+// TestUpstreamStatesUnsetSourceIsNil pins the honest degrade: `carlos edge`
+// and every test that never calls setUpstreamSource must read nil, not panic
+// — statusLoop calls upstreamStates() on every tick regardless of whether the
+// edge-ready callback has fired yet (it runs its first tick a minute in, but
+// the ordering is not something this loop gets to assume).
+func TestUpstreamStatesUnsetSourceIsNil(t *testing.T) {
+ setUpstreamSource(nil)
+ if got := upstreamStates(); got != nil {
+ t.Errorf("upstreamStates() with no source = %+v, want nil", got)
+ }
+}
+
// TestWrapListeningProbeNullsUnitBackedSockets is fix round 1's regression
// test: listeningProbe's raw /proc/net/unix read is confidently WRONG for
// systemd-socket-activated (registry.BackingUnit) routes, since the
diff --git a/internal/status/contract_test.go b/internal/status/contract_test.go
index 064b2c40e..2aac4256b 100644
--- a/internal/status/contract_test.go
+++ b/internal/status/contract_test.go
@@ -63,11 +63,22 @@ var allowedBoxKeys = map[string][]string{
// safe for exactly the same bucket-private reason: BuildFleet's
// field-by-field copy never names "metrics" at all, so no hostname here
// can reach FleetBox or fleet.json regardless of what this map holds.
+ // "metrics.upstreams" (multi-upstream plan, Task 6) is route_versions'
+ // sibling and then some: its KEYS are route hostnames and its VALUES carry
+ // real backend addresses. It is safe on the identical structural ground —
+ // BuildFleet's field-by-field copy never names "metrics", so nothing under
+ // it can reach FleetBox or fleet.json — and it is spelled out to one more
+ // level here than route_versions needs to be, because its values are
+ // objects rather than plain strings: "metrics.upstreams.*[]" pins the
+ // per-upstream fields, so a future field on UpstreamState is a deliberate
+ // publication decision too, not a silent ride-along.
"": {"generated_at", "label", "region", "routes", "load", "capacity", "replication", "adoption_running", "config_path_running", "restart_path_running", "expired_canaries", "metrics"},
"routes": {"static", "resident", "hibernating", "awake", "asleep"},
"replication": {"healthy", "lag_seconds"},
- "metrics": {"mem_total_bytes", "mem_avail_bytes", "load1", "disk_free_bytes", "release", "tunnel", "route_versions"},
+ "metrics": {"mem_total_bytes", "mem_avail_bytes", "load1", "disk_free_bytes", "release", "tunnel", "route_versions", "upstreams"},
"metrics.route_versions": {wildcardAnyKey},
+ "metrics.upstreams": {wildcardAnyKey},
+ "metrics.upstreams.*[]": {"addr", "up", "since", "fails", "last_error"},
}
// allowedFleetKeys are THE schema for Fleet's JSON output. Same friction as
@@ -99,12 +110,14 @@ func assertOnlyKeys(t *testing.T, j []byte, allowedKeys map[string][]string) {
// wildcardAnyKey marks a path (in an allowedKeys map) whose object keys are
// DATA, not a fixed field-name schema — metrics.route_versions (Task 8,
-// issue #78) is host -> Version, so its keys are hostnames chosen at
+// issue #78) is host -> Version and metrics.upstreams (multi-upstream plan,
+// Task 6) is host -> []UpstreamState, so their keys are hostnames chosen at
// runtime, which no fixed allowlist could name in advance. Values still
-// recurse normally (a route_versions entry that ever held a nested object
-// instead of a plain string would still trip the "unexpected key" check one
-// level down), so this only relaxes the KEY-NAME check, never the shape
-// check.
+// recurse normally, so this only relaxes the KEY-NAME check, never the shape
+// check: a wildcarded map's values walk on under the literal path segment
+// "*" (walkKeys, below), which is what lets "metrics.upstreams.*[]" pin the
+// fields of the objects inside — a runtime-chosen key must not also buy its
+// value an unexamined shape.
const wildcardAnyKey = "*"
// walkKeys recursively traverses the decoded JSON structure, checking that
@@ -133,11 +146,19 @@ func walkKeys(t *testing.T, obj interface{}, path string, allowedKeys map[string
continue
}
}
- // Recurse into this key's value, building the child path.
+ // Recurse into this key's value, building the child path. Under a
+ // wildcard the key itself is data (a hostname), so every value
+ // walks under the same literal "*" segment rather than a path
+ // only that one runtime key would ever match — otherwise the
+ // values of a wildcarded map could never be allowlisted at all,
+ // and any shape inside them would pass unchecked.
var childPath string
- if path == "" {
+ switch {
+ case wildcard:
+ childPath = path + ".*"
+ case path == "":
childPath = key
- } else {
+ default:
childPath = path + "." + key
}
walkKeys(t, v[key], childPath, allowedKeys)
@@ -256,9 +277,21 @@ func fixtureProbes() status.Probes {
return nil
}
+ // One probed route, so metrics.upstreams is actually exercised by the
+ // allowlist walk rather than merely permitted by it — an unexercised
+ // schema entry guards nothing. Distinct from dryrun.go's own upstream
+ // fixture for the same reason every value here is (see the doc comment).
+ upstreams := map[string][]status.UpstreamState{
+ "contract-fixture.internal": {
+ {Addr: "10.31.31.11:9090", Up: true, Since: time.Now().Add(-time.Hour)},
+ {Addr: "10.31.31.12:9090", Up: false, Since: time.Now().Add(-time.Minute), Fails: 5, LastError: "status 500 (want 200)"},
+ },
+ }
+
return status.Probes{
MachineID: "test-machine-id",
Region: "us-west-1",
+ Upstreams: upstreams,
MemFrac: &memFrac,
DiskFrac: &diskFrac,
Listening: listening,
diff --git a/internal/status/dryrun.go b/internal/status/dryrun.go
index aec3da27b..557002f3d 100644
--- a/internal/status/dryrun.go
+++ b/internal/status/dryrun.go
@@ -89,6 +89,22 @@ const (
// the identical private home: Box.Metrics.
var dryRunRouteVersionNeedles = []string{dryRunRouteVersionHost, dryRunRouteVersionVersion}
+// dryRunUpstreamHost and dryRunUpstreamAddr are the multi-upstream plan's own
+// needle pair (Task 6) for Box.Metrics.Upstreams — the probe table, which is
+// the most identifying thing this block carries: a route hostname as the KEY
+// and a real backend address inside the value. Distinct strings again from
+// every other needle set here (an address shape no other fixture uses, so a
+// digit collision with dryRunMetricsNeedles' 4242s is impossible), and scoped
+// the same way for the same reason: REQUIRED in status/boxes/*.json, so the
+// self-check can't pass by never populating the field, and BANNED everywhere
+// else DryRun publishes.
+const (
+ dryRunUpstreamHost = "upstream-secret.internal"
+ dryRunUpstreamAddr = "10.77.77.7:7717"
+)
+
+var dryRunUpstreamNeedles = []string{dryRunUpstreamHost, dryRunUpstreamAddr}
+
// ScanForLeaks reports which of banned's strings appear anywhere in j. It is
// the single implementation both DryRun's self-check and
// contract_test.go's scanForLeaks test helper run — the test package can't
@@ -170,6 +186,15 @@ func dryRunFixtures() ([]registry.Route, Probes) {
DiskFreeBytes: &diskFree,
Release: "dryrun4242sha",
Tunnel: "reconnecting",
+ // The probe table, keyed by a route host and carrying a backend
+ // address — see dryRunUpstreamNeedles. One up and one down upstream,
+ // so the down one's reason string is exercised too.
+ Upstreams: map[string][]UpstreamState{
+ dryRunUpstreamHost: {
+ {Addr: dryRunUpstreamAddr, Up: true, Since: time.Now().Add(-time.Hour)},
+ {Addr: "10.77.77.8:7717", Up: false, Since: time.Now().Add(-time.Minute), Fails: 3, LastError: "dial: connection refused"},
+ },
+ },
}
return routes, probes
}
@@ -279,6 +304,9 @@ func DryRun(dir string, out io.Writer) error {
// status/boxes/ object carrying the route_versions fixture, so this
// self-check can't pass by simply never exercising RouteVersions.
routeVersionKeySeen := false
+ // upstreamKeySeen is the same positive half for the probe table
+ // (dryRunUpstreamNeedles) — Box.Metrics.Upstreams, Task 6.
+ upstreamKeySeen := false
for _, key := range keys {
body, _, err := s.Get(ctx, key)
@@ -324,6 +352,20 @@ func DryRun(dir string, out io.Writer) error {
return fmt.Errorf("dry run: %s leaked a route_versions fixture %v — RouteVersions must never reach the public feed", key, leaks)
}
+ // dryRunUpstreamNeedles: the third scope split, on the same pattern.
+ // Upstreams is Box.Metrics's other hostname-keyed field and the only
+ // one carrying backend ADDRESSES, so status/boxes/*.json is its one
+ // privileged home and fleet.json/index.html must never show either
+ // half of the pair.
+ if strings.HasPrefix(key, "status/boxes/") {
+ if leaks := ScanForLeaks(body, dryRunUpstreamNeedles); len(leaks) == 0 {
+ return fmt.Errorf("dry run: %s: expected upstreams fixtures %v, found none — Upstreams not attached?", key, dryRunUpstreamNeedles)
+ }
+ upstreamKeySeen = true
+ } else if leaks := ScanForLeaks(body, dryRunUpstreamNeedles); len(leaks) > 0 {
+ return fmt.Errorf("dry run: %s leaked an upstreams fixture %v — Upstreams must never reach the public feed", key, leaks)
+ }
+
// Page assets (index.html, status.css) aren't JSON and simply fail
// to decode here — silently skipped, since the band-digit check
// only means anything for objects that actually carry Load/Capacity
@@ -342,6 +384,9 @@ func DryRun(dir string, out io.Writer) error {
if !routeVersionKeySeen {
return fmt.Errorf("dry run: no status/boxes/*.json object was scanned — route_versions scope check never ran")
}
+ if !upstreamKeySeen {
+ return fmt.Errorf("dry run: no status/boxes/*.json object was scanned — upstreams scope check never ran")
+ }
fmt.Fprintln(out, "dry-run: no leaks")
return nil
diff --git a/internal/status/snapshot.go b/internal/status/snapshot.go
index 39b18fc08..cb7fb3eb1 100644
--- a/internal/status/snapshot.go
+++ b/internal/status/snapshot.go
@@ -93,6 +93,43 @@ type Probes struct {
// unknown/not-applicable for both, never a guess.
Release string
Tunnel string
+ // Upstreams is this box's edge probe table — route host -> per-upstream
+ // state — read WHOLE from the prober on every tick (internal/edge's
+ // Edge.UpstreamStates), never accumulated across ticks. That rebuild is
+ // the field's staleness contract: a route the box has stopped probing
+ // (deleted, or its addrs cleared) simply isn't in the next tick's map, so
+ // no ghost entry can outlive the route that produced it. Empty/nil means
+ // this box probes nothing — the shape of every box with no multi-upstream
+ // route, which is most of them — and publishes as an absent key, not an
+ // empty object.
+ Upstreams map[string][]UpstreamState
+}
+
+// UpstreamState is one upstream's health as the edge's prober currently sees
+// it: what it believes, when that belief last changed, and why.
+//
+// It lives HERE, in the writer of the snapshot, rather than in internal/edge
+// where the prober that fills it lives, for one reason: this is a WIRE type on
+// status/boxes/<label>.json, and its readers are the console
+// (internal/console's box-detail views, which already parse BoxMetrics) and
+// operators. Declaring it in internal/edge would force every one of those
+// readers to import the edge — the whole ACME/DNS/proxy stack — just to name
+// the shape of a struct it decodes. internal/edge aliases this type instead
+// (type UpstreamState = status.UpstreamState), so the prober keeps writing
+// UpstreamState in its own package with no conversion at the boundary, and the
+// dependency runs edge -> status, the direction that leaves internal/status a
+// leaf over registry/release.
+//
+// Like everything else under Box.Metrics this is PRIVATE: Addr is a real
+// backend address and the map key is a real route host, safe here only because
+// BuildFleet never copies Metrics into the public fleet.json (see the package
+// doc comment).
+type UpstreamState struct {
+ Addr string `json:"addr"`
+ Up bool `json:"up"`
+ Since time.Time `json:"since"` // last transition (or the sync that started this prober)
+ Fails int `json:"fails"` // consecutive probe failures
+ LastError string `json:"last_error,omitempty"` // dial vs status, for the log/dashboard
}
// BoxMetrics is the raw, PRIVATE block: real numbers, never banded. It lives
@@ -118,6 +155,19 @@ type BoxMetrics struct {
// no versioned route, same "absent, not empty" rule as the rest of this
// block.
RouteVersions map[string]string `json:"route_versions,omitempty"`
+ // Upstreams is the box's per-upstream probe table, route host -> the
+ // UpstreamState of each addr behind it (multi-upstream routes only; a
+ // single-addr route with no health config is never probed and never
+ // appears). It is RouteVersions' twin in every way that matters here: a
+ // hostname-keyed map that belongs in this block and nowhere else, carried
+ // straight through from Probes.Upstreams with no banding, and omitted
+ // entirely — not present-and-empty — on a box that probes nothing, so
+ // every consumer written before this field parses such a box unchanged.
+ //
+ // Size is bounded by (probed routes × addrs per route) — a handful of
+ // entries on the busiest box today, and self-limiting: only routes an
+ // operator gave an addr list AND a health block are in it at all.
+ Upstreams map[string][]UpstreamState `json:"upstreams,omitempty"`
}
// RouteCounts is aggregate-only: no host, no socket path, just how many.
@@ -381,19 +431,29 @@ func CollectBox(routes []registry.Route, p Probes, now time.Time) Box {
ExpiredCanaries: p.ExpiredCanaries,
}
- // Metrics attaches only when something is known — an all-unknown Probes
- // AND no versioned route publishes no block at all, the same "unknown
- // stays absent, never a guess" rule every other field on Box follows,
- // and it keeps a box that can't answer any of the seven fields
- // indistinguishable on the wire from a box running code that predates
- // Metrics entirely.
+ // Metrics attaches only when something is known — an all-unknown Probes,
+ // no versioned route AND nothing probed publishes no block at all, the
+ // same "unknown stays absent, never a guess" rule every other field on
+ // Box follows, and it keeps a box that can't answer any of the eight
+ // fields indistinguishable on the wire from a box running code that
+ // predates Metrics entirely. An EMPTY (non-nil) probe table counts as
+ // nothing known for exactly the same reason an empty routeVersions map
+ // does: a box with no multi-upstream route has nothing to say, not
+ // something empty to say.
if p.MemTotalBytes != nil || p.MemAvailBytes != nil || p.Load1 != nil ||
- p.DiskFreeBytes != nil || p.Release != "" || p.Tunnel != "" || len(routeVersions) > 0 {
+ p.DiskFreeBytes != nil || p.Release != "" || p.Tunnel != "" ||
+ len(routeVersions) > 0 || len(p.Upstreams) > 0 {
box.Metrics = &BoxMetrics{
MemTotalBytes: p.MemTotalBytes, MemAvailBytes: p.MemAvailBytes,
Load1: p.Load1, DiskFreeBytes: p.DiskFreeBytes,
Release: p.Release, Tunnel: p.Tunnel,
RouteVersions: routeVersions,
+ // Carried through as the prober handed it over: proberSet.State
+ // already returns a fresh copy per call, so there is nothing here
+ // for a later probe tick to mutate under the marshaller. An empty
+ // map (rather than nil) reaching this line can only happen
+ // alongside another known field above, and omitempty drops it.
+ Upstreams: p.Upstreams,
}
}
diff --git a/internal/status/status_test.go b/internal/status/status_test.go
index 3d0a93e94..12dd4b08c 100644
--- a/internal/status/status_test.go
+++ b/internal/status/status_test.go
@@ -6,6 +6,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "reflect"
"strings"
"testing"
"time"
@@ -340,6 +341,84 @@ func TestCollectBoxRouteVersionsEmptyOmitsField(t *testing.T) {
}
}
+// ---- Task 6 (multi-upstream plan): per-upstream probe state ----------------
+
+// TestCollectBoxCarriesUpstreams is route_versions' sibling for the probe
+// table: CollectBox carries Probes.Upstreams onto the private Metrics block
+// verbatim, WriteBox lands it in status/boxes/<label>.json, and BuildFleet's
+// public fleet.json shows no trace of either the route host or the upstream
+// ADDRESS — the most identifying pair this block has ever carried, and safe
+// for exactly the same structural reason (BuildFleet never names Metrics).
+func TestCollectBoxCarriesUpstreams(t *testing.T) {
+ routes := []registry.Route{{Host: "ha.test", Kind: "service", Status: "ready"}}
+ since := time.Now().Add(-time.Minute).UTC().Truncate(time.Second)
+ p := status.Probes{MachineID: "m", Upstreams: map[string][]status.UpstreamState{
+ "ha.test": {
+ {Addr: "10.9.9.11:8080", Up: true, Since: since},
+ {Addr: "10.9.9.12:8080", Up: false, Since: since, Fails: 4, LastError: "dial: connection refused"},
+ },
+ }}
+
+ b := status.CollectBox(routes, p, time.Now())
+ if b.Metrics == nil {
+ t.Fatal("metrics block missing — Upstreams should have attached it")
+ }
+ if len(b.Metrics.Upstreams) != 1 || len(b.Metrics.Upstreams["ha.test"]) != 2 {
+ t.Fatalf("upstreams = %+v, want one host with two upstreams", b.Metrics.Upstreams)
+ }
+
+ s := teststore(t)
+ ctx := context.Background()
+ if err := status.WriteBox(ctx, s, b); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := s.Get(ctx, "status/boxes/"+b.Label+".json")
+ if err != nil {
+ t.Fatalf("Get written box: %v", err)
+ }
+ var round status.Box
+ if err := json.Unmarshal(raw, &round); err != nil {
+ t.Fatalf("unmarshal written box: %v", err)
+ }
+ got := round.Metrics.Upstreams["ha.test"]
+ if !reflect.DeepEqual(got, p.Upstreams["ha.test"]) {
+ t.Errorf("written upstreams = %+v, want %+v: %s", got, p.Upstreams["ha.test"], raw)
+ }
+
+ f, err := status.BuildFleet(ctx, s, routes, time.Now(), 5*time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fj, err := json.Marshal(f)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, needle := range []string{"ha.test", "10.9.9.11:8080", "connection refused"} {
+ if strings.Contains(string(fj), needle) {
+ t.Errorf("fleet.json leaked %q from the private upstream table: %s", needle, fj)
+ }
+ }
+}
+
+// TestCollectBoxUpstreamsEmptyOmitsField proves a box that probes nothing
+// contributes no upstreams key at all — the same "absent, not empty" rule
+// route_versions follows, which is what keeps every pre-existing snapshot
+// consumer parsing a box written by this build unchanged.
+func TestCollectBoxUpstreamsEmptyOmitsField(t *testing.T) {
+ routes := []registry.Route{{Host: "h.test", Kind: "instance", Status: "ready"}}
+ b := status.CollectBox(routes, status.Probes{MachineID: "m", Upstreams: map[string][]status.UpstreamState{}}, time.Now())
+ if b.Metrics != nil {
+ t.Fatalf("an empty probe table must not attach a metrics block, got %+v", b.Metrics)
+ }
+ raw, err := json.Marshal(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(string(raw), "upstreams") {
+ t.Errorf("box JSON carries an upstreams key with nothing probed: %s", raw)
+ }
+}
+
// ---- Task 3: fleet merge + publisher --------------------------------------
// teststore is a fresh DirStore over a scratch directory — every fleet test