amadan

carlos / platform Public

host: a file walk the time budget cut is not a measure, and the status tick walks only when self-serve moves are in play

measureFiles returns errMeasureBudget beside truncated when its budget (or
its context) cut the walk; the stop at twice the cap stays a measure. The
tick's walks share a budget of their own, from after the replica listing;
a host whose walk it never reaches, or cuts, reports Known false.
selfServeStatusProbes publishes the box id, the instance id and the lanes
always, and the per-host facts (the walks and the replica listing, under
the caps) only when the box's environment sets CARLOS_SELF_SERVE_MOVES=
offline or the box takes part in an offline-queued or offline-staging move
with pinned participants.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude a19010e579183f7229a03a9b98ca62372eaac3d0 parent f43bd4f
2 files changed, +185 −13
  • internal/host/movesoffline_reports.go +78 −12
  • internal/host/movesoffline_reports_test.go +107 −1
diff --git a/internal/host/movesoffline_reports.go b/internal/host/movesoffline_reports.go
index 37370b1de..febca050c 100644
--- a/internal/host/movesoffline_reports.go
+++ b/internal/host/movesoffline_reports.go
@@ -334,7 +334,9 @@ func (w *movesWorker) removeEmptyStrays(host string, rt *registry.Route) (remove
// without stopping at the cap: past capBytes it records overCap and goes
// on, until budget has passed or the count reaches twice capBytes, where it
// stops and records truncated. A missing files directory is an empty one.
-// An error is a walk that could not read part of the tree.
+// An error is a walk that could not read part of the tree, or
+// errMeasureBudget (with truncated) for one the budget cut short: only the
+// stop at twice the cap is a measure with truncated set.
func measureFiles(ctx context.Context, files, envPath string, budget time.Duration, capBytes int64) (bytes int64, entries int, overCap, truncated bool, err error) {
deadline := time.Now().Add(budget)
stop := errors.New("stop")
@@ -372,9 +374,9 @@ func measureFiles(ctx context.Context, files, envPath string, budget time.Durati
if err != nil {
return err
}
- if ctx.Err() != nil || time.Now().After(deadline) {
+ if ctx.Err() != nil || !time.Now().Before(deadline) {
truncated = true
- return stop
+ return errMeasureBudget
}
if p == files {
return nil
@@ -417,6 +419,20 @@ type replicaListing struct {
// beside each host's own file budget.
const hostFactsBudget = 2 * time.Minute
+// hostFactsFilesBudget bounds one tick's file walks, all hosts together. A
+// variable so a test can spend it.
+var hostFactsFilesBudget = hostFactsBudget
+
+// errMeasureBudget is measureFiles' error for a walk its time budget cut
+// short: what it counted is not a measure.
+var errMeasureBudget = errors.New("the time budget ran out before the files were counted")
+
+// selfServeMovesEnv is the self-serve switch (the console's
+// CARLOS_SELF_SERVE_MOVES): set to "offline" in a box's environment, the
+// status tick collects every instance's facts whether or not a move names
+// the box.
+const selfServeMovesEnv = "CARLOS_SELF_SERVE_MOVES"
+
// hostFacts is the status tick's per-host block (status.HostFacts) for
// every instance row this box holds: the row's lifetime, the replica's
// bytes and objects from one listing of the box's replica prefix per tick,
@@ -441,6 +457,9 @@ func (w *movesWorker) hostFacts(ctx context.Context, routes []registry.Route, no
ctx, cancel := context.WithTimeout(ctx, hostFactsBudget)
defer cancel()
listing, listErr := w.listReplicas(ctx, now)
+ // The walks share one budget of their own, from after the listing: a
+ // host whose walk it never reaches, or cuts short, is not measured.
+ filesDeadline := time.Now().Add(hostFactsFilesBudget)
out := make(map[string]status.HostFacts, len(rows))
for _, rt := range rows {
f := status.HostFacts{CollectedAt: now.Unix(), Incarnation: rt.Incarnation}
@@ -456,14 +475,18 @@ func (w *movesWorker) hostFacts(ctx context.Context, routes []registry.Route, no
if err != nil {
errs = append(errs, "files: "+err.Error())
} else {
- budget := precheckFilesBudget
+ budget := min(precheckFilesBudget, time.Until(filesDeadline))
if dl, ok := ctx.Deadline(); ok && time.Until(dl) < budget {
budget = time.Until(dl)
}
- var merr error
- f.FilesBytes, f.FilesEntries, f.OverCap, f.Truncated, merr = measureFiles(ctx, files, envPath, budget, archive.Cap)
- if merr != nil {
- errs = append(errs, "files: "+merr.Error())
+ if budget <= 0 {
+ errs = append(errs, "files: not measured this tick: the time budget ran out before this host's walk")
+ } else {
+ var merr error
+ f.FilesBytes, f.FilesEntries, f.OverCap, f.Truncated, merr = measureFiles(ctx, files, envPath, budget, archive.Cap)
+ if merr != nil {
+ errs = append(errs, "files: "+merr.Error())
+ }
}
sort.Strings(stray)
f.Strays = stray
@@ -537,13 +560,56 @@ func moveLanes(getenv func(string) string) []string {
}
// selfServeStatusProbes fills the status tick's self-serve block: the box id
-// the edges writer publishes, the lanes, and, from the armed moves worker,
-// the per-host facts. A box with no armed worker publishes the id and the
-// lanes alone.
+// the edges writer publishes, the instance id (the tick's own, else the
+// worker's lazily resolved one) and the lanes, always; and, from the armed
+// moves worker, the per-host facts, only when self-serve moves are in play
+// for this box (selfServeInPlay). The tick runs this before the heartbeat,
+// so a box no self-serve move can use pays nothing for the walks and the
+// replica listing.
func selfServeStatusProbes(ctx context.Context, p *status.Probes, w *movesWorker, boxID string, getenv func(string) string, routes []registry.Route, now time.Time) {
p.BoxID = boxID
p.Lanes = moveLanes(getenv)
- if w != nil {
+ if w == nil {
+ return
+ }
+ if p.InstanceID == "" {
+ p.InstanceID = w.cfg.instance()
+ }
+ if selfServeInPlay(w, getenv) {
p.Hosts = w.hostFacts(ctx, routes, now)
}
}
+
+// selfServeInPlay is whether the status tick collects the per-host facts:
+// the box's environment turns the self-serve switch on
+// (CARLOS_SELF_SERVE_MOVES=offline, the console's own switch, set on the
+// box by the deployment's configuration), or the box takes part in an
+// offline-queued or offline-staging move with pinned participants, as the
+// keep view's last listing read the records. A view whose listing failed
+// counts as taking part: the walks are capped, and a source box's facts
+// must not go missing because one listing did.
+func selfServeInPlay(w *movesWorker, getenv func(string) string) bool {
+ if strings.TrimSpace(getenv(selfServeMovesEnv)) == "offline" {
+ return true
+ }
+ v := w.cfg.Offline
+ if v == nil {
+ return false
+ }
+ snap := v.snap.Load()
+ if snap == nil {
+ return false
+ }
+ if snap.failed {
+ return true
+ }
+ for _, m := range snap.moves {
+ if m.Participants == nil || (m.State != provision.OfflineQueued && m.State != provision.OfflineStaging) {
+ continue
+ }
+ if source, target, _, _ := v.box.offlineRoles(&m); source || target {
+ return true
+ }
+ }
+ return false
+}
diff --git a/internal/host/movesoffline_reports_test.go b/internal/host/movesoffline_reports_test.go
index 07c6fcdbe..c6cf3e883 100644
--- a/internal/host/movesoffline_reports_test.go
+++ b/internal/host/movesoffline_reports_test.go
@@ -985,8 +985,11 @@ func TestStatusHostsBlockThroughTheRealPath(t *testing.T) {
now := time.Now()
p := status.Probes{MachineID: "machine-ie", InstanceID: ssSourceInstance, Release: "v1"}
getenv := func(k string) string {
- if k == moveLanesEnv {
+ switch k {
+ case moveLanesEnv:
return "data-uk, data-fr data-uk"
+ case selfServeMovesEnv:
+ return "offline"
}
return ""
}
@@ -1182,3 +1185,106 @@ func TestThrottledInstanceID(t *testing.T) {
t.Fatal("a known id was asked again")
}
}
+
+// A walk the time budget cut short, or never let start, is not a measure:
+// the host's facts say they are not known (Tasks 5 and 7 read "could not
+// measure"). A walk stopped at twice the cap is a measure: over the cap.
+func TestHostFactsUnknownWhenTheWalkIsCut(t *testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+ files := filepath.Join(dir, "files")
+ writeFile(t, filepath.Join(files, "a"), "0123456789")
+ if _, _, _, trunc, err := measureFiles(ctx, files, "", 0, 100); !trunc || !errors.Is(err, errMeasureBudget) {
+ t.Fatalf("a spent budget: truncated %v, err %v", trunc, err)
+ }
+ if _, _, over, trunc, err := measureFiles(ctx, files, "", time.Minute, 4); !over || !trunc || err != nil {
+ t.Fatalf("twice the cap: over %v truncated %v err %v", over, trunc, err)
+ }
+
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ routes := []registry.Route{*ie.row(offHost)}
+ saved := hostFactsFilesBudget
+ t.Cleanup(func() { hostFactsFilesBudget = saved })
+ hostFactsFilesBudget = 0
+ f := ie.w.hostFacts(ctx, routes, time.Now())[offHost]
+ if f.Known || !strings.Contains(f.Err, "not measured") || f.ReplicaCollectedAt == 0 {
+ t.Fatalf("a host whose walk never started: %+v", f)
+ }
+ hostFactsFilesBudget = saved
+ if f := ie.w.hostFacts(ctx, routes, time.Now())[offHost]; !f.Known {
+ t.Fatalf("with the budget back: %+v", f)
+ }
+}
+
+// The status tick walks files and lists the replica only on a box that
+// takes part in a queued or staging self-serve move, or whose deployment
+// turned the self-serve switch on (CARLOS_SELF_SERVE_MOVES=offline in the
+// agent's environment); the box id, instance id and lanes are published
+// always.
+func TestStatusBlockWalksOnlyWhenSelfServeIsInPlay(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ ieB := sw.sourceBox("ie-b")
+ ssIdentity(ieB, "i-ie-b")
+ sw.instance(ie)
+ ieB.tick()
+ ctx := context.Background()
+ lists := 0
+ sw.fake.FailOn = func(op, key string) error {
+ if op == "list" && strings.HasPrefix(key, sealPrefix) {
+ lists++
+ }
+ return nil
+ }
+ t.Cleanup(func() { sw.fake.FailOn = nil })
+ env := func(on bool) func(string) string {
+ return func(k string) string {
+ switch {
+ case k == moveLanesEnv:
+ return "data-uk"
+ case k == selfServeMovesEnv && on:
+ return "offline"
+ }
+ return ""
+ }
+ }
+ probe := func(b *offBox, on bool) status.Probes {
+ t.Helper()
+ routes, err := b.reg.List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ refreshOfflineView(ctx, sw.s, b.w.cfg.Offline)
+ var p status.Probes
+ selfServeStatusProbes(ctx, &p, b.w, b.name, env(on), routes, time.Now())
+ if p.BoxID != b.name || p.InstanceID != b.w.cfg.instance() || p.InstanceID == "" || !reflect.DeepEqual(p.Lanes, []string{"data-uk"}) {
+ t.Fatalf("the block's identity: box %q instance %q lanes %v", p.BoxID, p.InstanceID, p.Lanes)
+ }
+ return p
+ }
+ // No move, switch off: no walk, no listing.
+ if p := probe(ie, false); p.Hosts != nil || lists != 0 {
+ t.Fatalf("idle box, switch off: hosts %v, %d listings", p.Hosts, lists)
+ }
+ // The switch on: the facts, under the caps.
+ if p := probe(ie, true); !p.Hosts[offHost].Known || lists == 0 {
+ t.Fatalf("switch on: hosts %v, %d listings", p.Hosts, lists)
+ }
+ // Switch off, a queued self-serve move: its participant collects, a box
+ // it does not name does not.
+ sw.queue()
+ lists = 0
+ if p := probe(ie, false); !p.Hosts[offHost].Known || lists == 0 {
+ t.Fatalf("participant of a queued move: hosts %v, %d listings", p.Hosts, lists)
+ }
+ lists = 0
+ if p := probe(ieB, false); p.Hosts != nil || lists != 0 {
+ t.Fatalf("a box the move does not name: hosts %v, %d listings", p.Hosts, lists)
+ }
+ sw.toStaging()
+ if p := probe(ie, false); p.Hosts == nil {
+ t.Fatal("participant of a staging move collected nothing")
+ }
+}