amadan

carlos / platform Public

console: a rate term's rate is measured past its fixed part (bytes / max(seconds - fixed, 100 ms) over rows of at least 1 MiB), and rows under the floor measure the fixed part, so small final copies no longer predict a large replica as hours

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 8082a2280bbc90dcf53bc81d8c40127cd53ac6dc parent a1ec926
3 files changed, +134 −51
  • internal/console/selfserve_estimate.go +71 −37
  • internal/console/selfserve_estimate_test.go +60 −11
  • internal/e2e/selfserve/cut_test.go +3 −3
diff --git a/internal/console/selfserve_estimate.go b/internal/console/selfserve_estimate.go
index 8a3b67a4a..d05799c8e 100644
--- a/internal/console/selfserve_estimate.go
+++ b/internal/console/selfserve_estimate.go
@@ -20,10 +20,10 @@ import (
//
// Each term starts at its default and, once the region pair has five
// finished moves' timings for it, is refined from the median of the last
-// twenty (Task 9): a fixed term (stop, open) takes the median of the
-// phase's seconds; a rate term (copy, land, promote) the median of
-// bytes/seconds over the rows that carried bytes, and keeps its default
-// fixed part. The timings are the marks the console observed (console
+// twenty (Task 9): every term's fixed part from the rows under 1 MiB (the
+// median of their seconds), and a rate term's (copy, land, promote) rate
+// from the rows of at least 1 MiB, measured past that fixed part
+// (estimateRules). The timings are the marks the console observed (console
// observation times, insert-only, never overwritten) and the five rows a
// finished move derives from them; a move that aborted, or lacks a mark,
// contributes no row for that phase.
@@ -68,50 +68,70 @@ const (
estimateMinRows = 5
)
+// estimateRateFloor is the least a timing row must carry to measure a
+// rate; a row under it measures its term's fixed part only.
+const estimateRateFloor = mib
+
+// estimateRateEpsilon bounds a rate sample's time past the fixed part
+// from below: a row whose seconds are all overhead (a small final copy)
+// says its bytes moved in at most this long, never in no time.
+const estimateRateEpsilon = 100 * time.Millisecond
+
// estimateRules are the terms' rules for the region pair from -> to: the
-// defaults, each refined from the pair's timings when it has enough. A
-// timing that cannot be read keeps the default (logged): the estimate is
-// never refused for it.
+// defaults, each refined from the pair's timings when it has enough. For
+// every term, the rows under estimateRateFloor measure its fixed part: the
+// median of their seconds once there are estimateMinRows of the last
+// estimateWindow, else the default. For a rate term (copy, land, promote)
+// the rows at or over the floor measure its rate past that fixed part:
+// the median of bytes / max(seconds - fixed, estimateRateEpsilon), once
+// there are enough, else the default. So a phase's overhead is counted
+// once, in the fixed part, and never again in the rate. A timing that
+// cannot be read keeps the default (logged): the estimate is never refused
+// for it.
func (s *Server) estimateRules(from, to string) struct{ Stop, Copy, Land, Open, Promote term } {
r := estimateDefaults
if s == nil || s.store == nil {
return r
}
- fixed := func(phase string, t *term) {
- rows, err := s.store.MoveTimings(from, to, phase, false, estimateWindow)
+ window := func(phase string, rated bool) []MoveTiming {
+ rows, err := s.store.moveTimingWindow(from, to, phase, rated, estimateWindow)
if err != nil {
log.Printf("console: the move estimate: the %s timings for %s -> %s: %v (the default stands)", phase, from, to, err)
- return
+ return nil
}
if len(rows) < estimateMinRows {
- return
- }
- xs := make([]float64, len(rows))
- for i, m := range rows {
- xs[i] = m.Seconds
+ return nil
}
- t.Fixed = time.Duration(median(xs) * float64(time.Second))
+ return rows
}
- rate := func(phase string, t *term) {
- rows, err := s.store.MoveTimings(from, to, phase, true, estimateWindow)
- if err != nil {
- log.Printf("console: the move estimate: the %s timings for %s -> %s: %v (the default stands)", phase, from, to, err)
- return
+ refine := func(phase string, t *term) {
+ if rows := window(phase, false); rows != nil {
+ xs := make([]float64, len(rows))
+ for i, m := range rows {
+ xs[i] = m.Seconds
+ }
+ t.Fixed = time.Duration(median(xs) * float64(time.Second))
}
- if len(rows) < estimateMinRows {
- return
+ if t.BytesPerSecond == 0 {
+ return // a fixed term
}
- xs := make([]float64, len(rows))
- for i, m := range rows {
- xs[i] = float64(m.Bytes) / m.Seconds
+ if rows := window(phase, true); rows != nil {
+ xs := make([]float64, len(rows))
+ for i, m := range rows {
+ past := m.Seconds - t.Fixed.Seconds()
+ if past < estimateRateEpsilon.Seconds() {
+ past = estimateRateEpsilon.Seconds()
+ }
+ xs[i] = float64(m.Bytes) / past
+ }
+ t.BytesPerSecond = median(xs)
}
- t.BytesPerSecond = median(xs)
}
- fixed("stop", &r.Stop)
- rate("copy", &r.Copy)
- rate("land", &r.Land)
- fixed("open", &r.Open)
- rate("promote", &r.Promote)
+ refine("stop", &r.Stop)
+ refine("copy", &r.Copy)
+ refine("land", &r.Land)
+ refine("open", &r.Open)
+ refine("promote", &r.Promote)
return r
}
@@ -248,12 +268,26 @@ VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (id, phase) DO NOTHING`,
}
// MoveTimings are the region pair's newest rows for phase, at most limit,
-// newest first; withBytes keeps only the rows that carried bytes (a rate's
-// rows).
-func (s *Store) MoveTimings(from, to, phase string, withBytes bool, limit int) ([]MoveTiming, error) {
+// newest first.
+func (s *Store) MoveTimings(from, to, phase string, limit int) ([]MoveTiming, error) {
+ return s.moveTimings(`AND seconds >= 0`, from, to, phase, limit)
+}
+
+// moveTimingWindow is the estimate's window of a phase's rows: the rated
+// rows (at least estimateRateFloor bytes, a positive time) or the rows
+// under the floor, which measure the term's fixed part; newest first, at
+// most limit.
+func (s *Store) moveTimingWindow(from, to, phase string, rated bool, limit int) ([]MoveTiming, error) {
+ if rated {
+ return s.moveTimings(fmt.Sprintf(`AND bytes >= %d AND seconds > 0`, int64(estimateRateFloor)), from, to, phase, limit)
+ }
+ return s.moveTimings(fmt.Sprintf(`AND bytes < %d AND seconds >= 0`, int64(estimateRateFloor)), from, to, phase, limit)
+}
+
+func (s *Store) moveTimings(cond, from, to, phase string, limit int) ([]MoveTiming, error) {
rows, err := s.db.Query(`SELECT id, phase, from_region, to_region, bytes, seconds, at FROM move_timings
-WHERE from_region = ? AND to_region = ? AND phase = ? AND seconds >= 0 AND (? = 0 OR (bytes > 0 AND seconds > 0))
-ORDER BY at DESC, id DESC LIMIT ?`, from, to, phase, ssBool(withBytes), limit)
+WHERE from_region = ? AND to_region = ? AND phase = ? `+cond+`
+ORDER BY at DESC, id DESC LIMIT ?`, from, to, phase, limit)
if err != nil {
return nil, err
}
diff --git a/internal/console/selfserve_estimate_test.go b/internal/console/selfserve_estimate_test.go
index 28b01e671..134d48448 100644
--- a/internal/console/selfserve_estimate_test.go
+++ b/internal/console/selfserve_estimate_test.go
@@ -70,9 +70,11 @@ func TestEstimateUsesDefaultsThenMedians(t *testing.T) {
id := fmt.Sprintf("mv_estimate%02d", n)
ssTiming(t, st, id, "stop", 0, stop, at)
ssTiming(t, st, id, "open", 0, open, at)
- ssTiming(t, st, id, "copy", 100*mib, 100*mib/copyRate, at)
- ssTiming(t, st, id, "land", 110*mib, 110*mib/landRate, at)
- ssTiming(t, st, id, "promote", 100*mib, 100*mib/promoteRate, at)
+ // Each rate phase's seconds carry the default fixed part, as a
+ // real phase's do: seconds = fixed + bytes/rate.
+ ssTiming(t, st, id, "copy", 100*mib, 10+100*mib/copyRate, at)
+ ssTiming(t, st, id, "land", 110*mib, 30+110*mib/landRate, at)
+ ssTiming(t, st, id, "promote", 100*mib, 10+100*mib/promoteRate, at)
}
for i := 0; i < 4; i++ {
addMove(i, 60, 90, 10*mib, 5*mib, 2*mib)
@@ -94,13 +96,21 @@ func TestEstimateUsesDefaultsThenMedians(t *testing.T) {
check(t, "uk", "ie", defaults)
})
- // A rate row without bytes is no rate: twenty newer copy rows carrying
- // no bytes change nothing.
+ // A row under the rate floor (1 MiB) is no rate: it measures the
+ // term's fixed part. Twenty newer copies of half a MiB at 4 s make the
+ // copy's fixed part 4 s, and its rate is then measured past those 4 s
+ // (100 MiB in 20 - 4 s), so 100 MiB still takes 20 s.
for i := 0; i < 20; i++ {
at = at.Add(time.Minute)
- ssTiming(t, st, fmt.Sprintf("mv_nobytes%02d", i), "copy", 0, 1, at)
+ ssTiming(t, st, fmt.Sprintf("mv_nobytes%02d", i), "copy", mib/2, 4, at)
}
- t.Run("rows without bytes are not rates", func(t *testing.T) { check(t, "ie", "uk", medians) })
+ t.Run("rows under the floor measure the fixed part", func(t *testing.T) {
+ check(t, "ie", "uk", medians)
+ terms := s.estimateRules("ie", "uk")
+ if terms.Copy.Fixed != 4*time.Second || ssSeconds(time.Duration(float64(100*mib)/terms.Copy.BytesPerSecond*float64(time.Second))) != 16 {
+ t.Fatalf("copy rule %+v, want 4 s fixed and 100 MiB in 16 s", terms.Copy)
+ }
+ })
// The last twenty only: twenty newer moves move every median.
for i := 0; i < 20; i++ {
@@ -154,7 +164,8 @@ func TestEstimateBeforeAndAfterStaging(t *testing.T) {
s := &Server{store: st}
const replica, files = 3 << 30, 4 * mib
for i := 0; i < 5; i++ {
- ssTiming(t, st, fmt.Sprintf("mv_lane%02d", i), "copy", 100*mib, 2, ssT0.Add(time.Duration(i)*time.Minute))
+ // 100 MiB past the default 10 s fixed part in 2 s: 50 MiB/s.
+ ssTiming(t, st, fmt.Sprintf("mv_lane%02d", i), "copy", 100*mib, 12, ssT0.Add(time.Duration(i)*time.Minute))
}
// Before a move is requested the final copy is the whole replica (the
// door's estimate).
@@ -192,6 +203,44 @@ func TestEstimateBeforeAndAfterStaging(t *testing.T) {
}
}
+// TestEstimateRatesExcludeTheFixedPart: a rate is measured past the
+// term's fixed part, bytes / max(seconds - fixed, 100 ms), over rows of at
+// least 1 MiB; so small final copies, whose seconds are almost all
+// overhead, never predict a large replica as hours.
+func TestEstimateRatesExcludeTheFixedPart(t *testing.T) {
+ const replica = 3 << 30
+ pure := 10*time.Second + ssRate(replica, 25*mib) // the default copy term: about two minutes
+
+ t.Run("five 2 MiB final copies of about 5 s", func(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ for i, sec := range []float64{4.8, 4.9, 5, 5.1, 5.2} {
+ ssTiming(t, st, fmt.Sprintf("mv_small%02d", i), "copy", 2*mib, sec, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ got, _ := s.estimate("ie", "uk", replica, 0, 0)
+ if got.Copy > 2*pure || got.Copy < pure/2 {
+ t.Fatalf("a 3 GiB copy estimates %v from five small final copies, want within 2x of %v", got.Copy, pure)
+ }
+ })
+
+ t.Run("a measured fixed part and rate are recovered", func(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ for i := 0; i < 5; i++ {
+ at := ssT0.Add(time.Duration(i) * time.Minute)
+ ssTiming(t, st, fmt.Sprintf("mv_fixed%02d", i), "copy", mib/4, 6, at) // the overhead alone
+ ssTiming(t, st, fmt.Sprintf("mv_rated%02d", i), "copy", 200*mib, 6+200.0/40, at) // 6 s + 200 MiB at 40 MiB/s
+ }
+ r := s.estimateRules("ie", "uk")
+ if r.Copy.Fixed != 6*time.Second || r.Copy.BytesPerSecond != 40*mib {
+ t.Fatalf("copy rule %+v, want 6 s and 40 MiB/s", r.Copy)
+ }
+ if got, _ := s.estimate("ie", "uk", replica, 0, 0); ssSeconds(got.Copy) != ssSeconds(6*time.Second+ssRate(replica, 40*mib)) {
+ t.Fatalf("copy %v", got.Copy)
+ }
+ })
+}
+
// ssMarks writes every mark for id, a second apart from at, with the copy,
// landing and replica bytes.
func ssMarks(t *testing.T, st *Store, id string, at time.Time, skip ...string) {
@@ -297,7 +346,7 @@ func TestAbortedMovesContributeNoRates(t *testing.T) {
return row
}
finish("mv_abortedabortedabortedabort", "a."+ssHost, provision.OfflineOutcomeAborted)
- if got, err := st.MoveTimings("ie", "uk", "stop", false, 20); err != nil || len(got) != 0 {
+ if got, err := st.MoveTimings("ie", "uk", "stop", 20); err != nil || len(got) != 0 {
t.Fatalf("an aborted move left timings: %+v, %v", got, err)
}
@@ -312,7 +361,7 @@ func TestAbortedMovesContributeNoRates(t *testing.T) {
"promote": {Seconds: 32, Bytes: 100 * mib},
}
for phase, w := range want {
- got, err := st.MoveTimings("ie", "uk", phase, false, 20)
+ got, err := st.MoveTimings("ie", "uk", phase, 20)
if err != nil || len(got) != 1 {
t.Fatalf("%s: %+v, %v; want one row", phase, got, err)
}
@@ -324,7 +373,7 @@ func TestAbortedMovesContributeNoRates(t *testing.T) {
// A done move without a mark contributes no row for its phase.
finish("mv_nofirstoknofirstoknofirsto", "c."+ssHost, provision.OfflineOutcomeDone, "first_ok_at", "seal_at")
for phase, n := range map[string]int{"promote": 1, "copy": 1, "stop": 2, "land": 2, "open": 2} {
- if got, _ := st.MoveTimings("ie", "uk", phase, false, 20); len(got) != n {
+ if got, _ := st.MoveTimings("ie", "uk", phase, 20); len(got) != n {
t.Errorf("%s: %d rows, want %d", phase, len(got), n)
}
}
diff --git a/internal/e2e/selfserve/cut_test.go b/internal/e2e/selfserve/cut_test.go
index 489fd6605..5ab625eda 100644
--- a/internal/e2e/selfserve/cut_test.go
+++ b/internal/e2e/selfserve/cut_test.go
@@ -175,13 +175,13 @@ func TestCutRecordsTimingsAndDeliversEvents(t *testing.T) {
t.Fatalf("landed_at %d bytes, landed_replica %d: want the files and the replica, and the replica", marks["landed_at"].Bytes, marks["landed_replica"].Bytes)
}
for _, phase := range []string{"stop", "copy", "land", "open", "promote"} {
- got, err := w.cst.MoveTimings("", host.E2ETargetRegion, phase, false, 20)
+ got, err := w.cst.MoveTimings("", host.E2ETargetRegion, phase, 20)
if err != nil || len(got) != 1 || got[0].ID != row.ID {
t.Fatalf("%s timings %+v, %v; want the move's one row", phase, got, err)
}
}
- if got, _ := w.cst.MoveTimings("", host.E2ETargetRegion, "land", true, 20); len(got) != 1 || got[0].Bytes != marks["landed_at"].Bytes {
- t.Fatalf("land's rate row %+v, want the landed bytes %d", got, marks["landed_at"].Bytes)
+ if got, _ := w.cst.MoveTimings("", host.E2ETargetRegion, "land", 20); len(got) != 1 || got[0].Bytes != marks["landed_at"].Bytes {
+ t.Fatalf("land's row %+v, want the landed bytes %d", got, marks["landed_at"].Bytes)
}
// The terminal event is the last tick's: one more delivers it.
for i := 0; i < 3; i++ {