diff --git a/docs/cluster.md b/docs/cluster.md index f7e5a715..35a533ed 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -450,9 +450,22 @@ the rounds reached, plus 400 ms for the receive and the start, fits never fit), the source pauses the guest for the final round and the target receives the series and starts the guest. A move whose deltas never fit answers 503 and the guest runs on; from the second delta on, one -that shrinks by under 10% and is still too large answers at once. `downtime_ms` in the answer +that shrinks by under 10% and is still too large answers at once. The estimate +cannot see the final capture, which runs with the guest paused, or the pages +dirtied since the last delta, so once the guest is paused the source checks again: the final +capture's time plus the final round's bytes on the wire (the last delta's, plus +the growth of the final round) at the rounds' rate plus the 400 ms. Past `max_downtime_ms` it aborts the +move before sending the final round: `cocoon vm migrate abort` resumes the guest +on the source, the target drops what it received, the record ends `aborted`, and +the move answers 503. The check cannot give back a pause already spent, so a final +capture that alone overruns the budget still pauses the guest for the capture plus +the abort and its record write; with the meta store up, that, not +`max_downtime_ms`, is the worst case. Past the check, a slow +transfer or commit still lengthens the pause, and `downtime_ms` reports it. `downtime_ms` in the answer is what the source measured from the final capture's start, including its wait -for a free dump slot, until the target answers the commit, after it started the guest. Each round has one minute plus its size at 10 MB/s to arrive, so a +for a free dump slot, until the target answers the commit, after it started the guest; +of which `capture_ms`, `send_ms` and `commit_ms` are the final capture, the final +round's transfer and the commit. Each round has one minute plus its size at 10 MB/s to arrive, so a target that stops reading fails the move. A replicated claim's registry unrow, the final capture, its transfer and the target's commit have 2 minutes together; past that the source resolves the move through the record, and the guest resumes on the source unless the target committed. diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 596484f6..75cc628e 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -914,10 +914,10 @@ its guest's processes and memory (see `{"to": "", "max_downtime_ms": 2000}`, `to` as `GET /v1/peers` lists it and never this node (else 400), `max_downtime_ms` optional with a default of 2000 and a floor: 400 or less answers 400 at once, because the target's receive -and start alone are allowed that long. Answers `{"owner_addr": "", "rounds": n, "downtime_ms": d}` once the +and start alone are allowed that long. Answers `{"owner_addr": "", "rounds": n, "downtime_ms": d, "capture_ms": c, "send_ms": s, "commit_ms": m}` once the peer took the claim and this node dropped it; a guest the peer cannot start is failed there. `downtime_ms` runs from the final capture's start, including its -wait for a free dump slot, until the peer answers the commit, after it started the guest. While the move runs, release, fork, +wait for a free dump slot, until the peer answers the commit, after it started the guest; of it, `capture_ms`, `send_ms` and `commit_ms` are the final capture, the final round transfer and the commit. A final round whose capture time and size no longer fit `max_downtime_ms` is aborted before it is sent and the guest resumes here, so with the meta store up a capture that alone overruns pauses the guest for the capture plus the abort and its record write. While the move runs, release, fork, checkpoint, promote, hibernate and volume changes of the claim answer 503 `sandbox moving to another node; retry`. Env writes, renew and data-plane calls keep working until the final round and answer the same 503 during it, because the peer would miss @@ -938,7 +938,8 @@ node drops the claim (see [moving a running claim](cluster.md#moving-a-running-c - 502 the target is unreachable or failed - 503 the move or archive registry is down, the node lease is overdue, the node is evacuating the claim (`sandbox moving to another node; retry`), or the - final round would not fit `max_downtime_ms`, after at most 4 deltas + final round would not fit `max_downtime_ms`, after at most 4 deltas or once + its capture overran it with the guest paused (the guest resumes here) The target serves root-only peer routes for a move, `POST /v1/moves/{id}`, `PUT /v1/moves/{id}/rounds/{n}`, `POST /v1/moves/{id}/commit` and diff --git a/sandboxd/pool/move.go b/sandboxd/pool/move.go index 8737d72d..706dc853 100644 --- a/sandboxd/pool/move.go +++ b/sandboxd/pool/move.go @@ -38,6 +38,9 @@ type MoveAdmission struct { type MoveResult struct { Rounds int Downtime time.Duration + Capture time.Duration + Send time.Duration + Commit time.Duration } // MovePeer reaches the target node of a live move. @@ -58,6 +61,9 @@ type outMove struct { final bool locked bool pausedAt time.Time + capture time.Duration + send time.Duration + commit time.Duration } func (mv *outMove) owns(rec moves.Record, node string) bool { @@ -65,7 +71,7 @@ func (mv *outMove) owns(rec moves.Record, node string) bool { } func (mv *outMove) result() MoveResult { - res := MoveResult{Rounds: mv.rounds} + res := MoveResult{Rounds: mv.rounds, Capture: mv.capture, Send: mv.send, Commit: mv.commit} if !mv.pausedAt.IsZero() { res.Downtime = time.Since(mv.pausedAt) } diff --git a/sandboxd/pool/move_out.go b/sandboxd/pool/move_out.go index 1f6354a5..50b9d464 100644 --- a/sandboxd/pool/move_out.go +++ b/sandboxd/pool/move_out.go @@ -74,13 +74,12 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b if err = m.journalMoveOut(ctx, sb, mv); err != nil { return err } - var sent, previous int64 + var size, sent, previous int64 var sending time.Duration for delta := 0; ; delta++ { start := time.Now() - size, sendErr := m.sendRound(ctx, sb.ID, mv, round) - if sendErr != nil { - return sendErr + if size, err = m.sendRound(ctx, sb.ID, mv, round); err != nil { + return err } sending += time.Since(start) sent += size @@ -97,6 +96,7 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b return fmt.Errorf("%w: delta capture: %w", ErrNotMovable, err) } } + lastBytes := roundBytes(round) m.mu.Lock() mv.final = true m.mu.Unlock() @@ -111,13 +111,21 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b if round, err = m.eng.MigrateCapture(ctx, vm, mv.Series, true); err != nil { return fmt.Errorf("%w: final capture: %w", ErrNotMovable, err) } + mv.capture = time.Since(mv.pausedAt) + apparent := roundBytes(round) + wire := size + max(apparent-lastBytes, 0) + if pause := mv.capture + finalPause(wire, sent, sending); pause > budget { + return fmt.Errorf("%w: the final round of %d bytes took %s to capture, leaving a pause of about %s", ErrMoveDowntime, apparent, mv.capture.Round(time.Millisecond), pause.Round(time.Millisecond)) + } final, err := m.moves.Advance(ctx, sb.ID, mv.Epoch, moves.Pending, moves.Final) if err != nil { return err } + start := time.Now() if _, err = m.sendRound(ctx, sb.ID, mv, round); err != nil { return err } + mv.send = time.Since(start) if used, ok := m.out.TunUsed(sb.ID); ok { m.setTunBytes(sb, used) } @@ -128,7 +136,10 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b if err != nil { return fmt.Errorf("encode claim: %w", err) } - return m.movePeer.Commit(ctx, mv.Addr, sb.ID, claim, final) + start = time.Now() + err = m.movePeer.Commit(ctx, mv.Addr, sb.ID, claim, final) + mv.commit = time.Since(start) + return err } func (m *Manager) beginMove(ctx context.Context, sb *types.Sandbox, to string) (int64, error) { @@ -155,11 +166,10 @@ func (m *Manager) beginMove(ctx context.Context, sb *types.Sandbox, to string) ( func (m *Manager) sendRound(ctx context.Context, id string, mv *outMove, round types.MigrateRound) (int64, error) { mv.rounds++ names := make([]string, len(round.Files)) - var size int64 for i, f := range round.Files { - names[i], size = f.Name, size+f.Size + names[i] = f.Name } - ctx, cancel := context.WithTimeout(ctx, moveRoundBase+time.Duration(size/moveRoundMinRate)*time.Second) + ctx, cancel := context.WithTimeout(ctx, moveRoundBase+time.Duration(roundBytes(round)/moveRoundMinRate)*time.Second) defer cancel() return m.movePeer.SendRound(ctx, mv.Addr, id, mv.rounds, round.Dir, names) } @@ -413,6 +423,14 @@ func (m *Manager) midMoveLocked(sb *types.Sandbox) bool { return moving || sb.MoveIn } +func roundBytes(round types.MigrateRound) int64 { + var size int64 + for _, f := range round.Files { + size += f.Size + } + return size +} + func finalPause(last, sent int64, sending time.Duration) time.Duration { return MoveStartAllowance + time.Duration(float64(sending)*float64(last)/float64(sent)) } diff --git a/sandboxd/pool/move_test.go b/sandboxd/pool/move_test.go index f4dbfeb6..3643cc8e 100644 --- a/sandboxd/pool/move_test.go +++ b/sandboxd/pool/move_test.go @@ -43,8 +43,8 @@ func TestMoveHandsARunningClaimToThePeerWithBothNodesCountingIt(t *testing.T) { if err != nil { t.Fatalf("move: %v", err) } - if res.Rounds != 3 || res.Downtime <= 0 { - t.Errorf("result %+v, want a baseline, one delta and the final round with a downtime", res) + if res.Rounds != 3 || res.Downtime <= 0 || res.Capture <= 0 || res.Send <= 0 || res.Commit <= 0 || res.Capture+res.Send+res.Commit > res.Downtime { + t.Errorf("result %+v, want a baseline, one delta and the final round with a downtime that covers its phases", res) } for i, counts := range during { if counts != [2][3]int{{1, 1, 1}, {1, 1, 1}} { @@ -741,6 +741,7 @@ func TestMoveGivesEachRoundADeadlineFromItsSize(t *testing.T) { mc := newMoveCluster(t) sb := mustClaim(t, mc.s, testKey) fakeOf(mc.s).migrateSizes = []int64{1 << 30, 1, 1 << 40} + mc.p.wireSizes = []int64{1 << 30, 0} if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Hour); err != nil { t.Fatalf("move: %v", err) } @@ -883,15 +884,78 @@ func TestMoveEstimatesThePauseFromTheBytesOnTheWire(t *testing.T) { if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+50*time.Millisecond); !errors.Is(err, ErrMoveDowntime) { t.Fatalf("move of a hot guest: %v, want ErrMoveDowntime", err) } - eng.mu.Lock() - wentFinal := slices.ContainsFunc(eng.captures, func(c string) bool { return strings.HasSuffix(c, ":true") }) - eng.mu.Unlock() - if wentFinal { + if finalCaptured(mc.s) { t.Error("the move paused the guest for a final round that cannot fit") } assertStillOnSource(t, mc, sb) } +func TestMoveWhosePausedRoundOverrunsTheBudgetResumesTheGuestOnTheSource(t *testing.T) { + for name, tt := range map[string]struct { + captureTime time.Duration + sizes, wire []int64 + }{ + "final capture": {captureTime: 200 * time.Millisecond}, + "final round grown": {sizes: []int64{1 << 20, 1 << 10, 1 << 40}, wire: []int64{1 << 20, 1 << 10}}, + } { + t.Run(name, func(t *testing.T) { + mc := newMoveCluster(t) + sb := mustClaim(t, mc.s, testKey) + eng := fakeOf(mc.s) + eng.finalCaptureTime, eng.migrateSizes, mc.p.wireSizes = tt.captureTime, tt.sizes, tt.wire + sentFinal, committed := false, false + mc.p.roundHook = func(round int) error { + sentFinal = sentFinal || round == 3 + return nil + } + mc.p.commitHook = func(int) error { committed = true; return nil } + if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+100*time.Millisecond); !errors.Is(err, ErrMoveDowntime) { + t.Fatalf("move: %v, want ErrMoveDowntime", err) + } + if !finalCaptured(mc.s) || sentFinal || committed { + t.Errorf("final capture %v sent %v commit %v, want the capture and neither the final round nor a commit", finalCaptured(mc.s), sentFinal, committed) + } + assertStillOnSource(t, mc, sb) + assertNoMoveResidue(t, mc.t, sb.ID) + if _, ok, _ := mc.moveReg.Get(t.Context(), sb.ID); ok { + t.Error("the aborted move left its record") + } + }) + } +} + +func TestMoveWhoseFinalRoundOutgrowsATinyDeltaStillFits(t *testing.T) { + mc := newMoveCluster(t) + sb := mustClaim(t, mc.s, testKey) + fakeOf(mc.s).migrateSizes = []int64{1 << 20, 1, 1 << 24} + mc.p.wireSizes = []int64{1 << 20, 1 << 12} + if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+500*time.Millisecond); err != nil { + t.Fatalf("move whose final round grew by 16 MiB over a 1-byte delta sent as 4 KiB: %v", err) + } +} + +func TestMoveWhoseFinalRoundShrinksStillCountsTheLastDeltasBytes(t *testing.T) { + mc := newMoveCluster(t) + sb := mustClaim(t, mc.s, testKey) + eng := fakeOf(mc.s) + eng.finalCaptureTime, eng.migrateSizes = 100*time.Millisecond, []int64{1 << 30, 512 << 20, 256 << 20} + mc.p.wireSizes = []int64{32 << 20, 32 << 20} + mc.p.roundHook = func(round int) error { + if round == 2 { + time.Sleep(200 * time.Millisecond) + } + return nil + } + if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+160*time.Millisecond); !errors.Is(err, ErrMoveDowntime) { + t.Fatalf("move whose sparse final round shrank after a slow 32 MiB delta: %v, want ErrMoveDowntime", err) + } + if !finalCaptured(mc.s) { + t.Error("the move refused before the final capture, want the paused-round check to refuse it") + } + assertStillOnSource(t, mc, sb) + assertNoMoveResidue(t, mc.t, sb.ID) +} + func TestMoveGivesUpOnDeltasThatStopShrinking(t *testing.T) { for name, tt := range map[string]struct { sizes []int64 @@ -1874,6 +1938,13 @@ func movingOn(m *Manager, sb *types.Sandbox) bool { return m.midMoveLocked(sb) } +func finalCaptured(m *Manager) bool { + eng := fakeOf(m) + eng.mu.Lock() + defer eng.mu.Unlock() + return slices.ContainsFunc(eng.captures, func(c string) bool { return strings.HasSuffix(c, ":true") }) +} + func pausedOn(m *Manager, vm string) bool { eng := fakeOf(m) eng.mu.Lock() diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 09f214e1..a3e3b451 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -1481,6 +1481,7 @@ type fakeEngine struct { migrateSupportErr error captureErr error finalCaptureErr error + finalCaptureTime time.Duration receiveErr error receiveStall chan struct{} receiveEntered chan struct{} @@ -2056,6 +2057,9 @@ func (f *fakeEngine) Balloon(_ context.Context, vmName, target string) (types.VM } func (f *fakeEngine) MigrateCapture(_ context.Context, vmName, series string, final bool) (types.MigrateRound, error) { + if final { + time.Sleep(f.finalCaptureTime) + } f.mu.Lock() defer f.mu.Unlock() f.captures = append(f.captures, fmt.Sprintf("%s:%s:%v", vmName, series, final)) diff --git a/sandboxd/server/move.go b/sandboxd/server/move.go index c60c19f0..ec382f4d 100644 --- a/sandboxd/server/move.go +++ b/sandboxd/server/move.go @@ -111,6 +111,9 @@ type MoveResponse struct { OwnerAddr string `json:"owner_addr"` Rounds int `json:"rounds"` DowntimeMS int64 `json:"downtime_ms"` + CaptureMS int64 `json:"capture_ms"` + SendMS int64 `json:"send_ms"` + CommitMS int64 `json:"commit_ms"` } func (s *Server) handleMove(w http.ResponseWriter, r *http.Request) { @@ -131,7 +134,10 @@ func (s *Server) handleMove(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") res, err := s.mgr.Move(r.Context(), id, addr, budget) writeResult(w, r, "move", id, "move failed", err, func() { - writeJSON(w, http.StatusOK, MoveResponse{OwnerAddr: s.placer.ClientAddr(addr), Rounds: res.Rounds, DowntimeMS: res.Downtime.Milliseconds()}) + writeJSON(w, http.StatusOK, MoveResponse{ + OwnerAddr: s.placer.ClientAddr(addr), Rounds: res.Rounds, DowntimeMS: res.Downtime.Milliseconds(), + CaptureMS: res.Capture.Milliseconds(), SendMS: res.Send.Milliseconds(), CommitMS: res.Commit.Milliseconds(), + }) }) } diff --git a/sandboxd/server/move_test.go b/sandboxd/server/move_test.go index cdd84cba..6bc6b78e 100644 --- a/sandboxd/server/move_test.go +++ b/sandboxd/server/move_test.go @@ -20,7 +20,7 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) { var outcome error mgr := &fakeManager{move: func(_, addr string, budget time.Duration) (pool.MoveResult, error) { gotAddr, gotBudget = addr, budget - return pool.MoveResult{Rounds: 3, Downtime: 180 * time.Millisecond}, outcome + return pool.MoveResult{Rounds: 3, Downtime: 180 * time.Millisecond, Capture: 60 * time.Millisecond, Send: 90 * time.Millisecond, Commit: 25 * time.Millisecond}, outcome }} srv := newMeshTestServer(t, "root", mgr, &fakePlacer{addrs: []string{"10.0.0.2:7777"}}, nil) @@ -30,7 +30,7 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) { if err := json.UnmarshalRead(resp.Body, &body); err != nil || resp.StatusCode != http.StatusOK { t.Fatalf("move: %d %v", resp.StatusCode, err) } - if gotAddr != "10.0.0.2:7777" || gotBudget != 1500*time.Millisecond || body != (MoveResponse{OwnerAddr: "10.0.0.2:7777", Rounds: 3, DowntimeMS: 180}) { + if gotAddr != "10.0.0.2:7777" || gotBudget != 1500*time.Millisecond || body != (MoveResponse{OwnerAddr: "10.0.0.2:7777", Rounds: 3, DowntimeMS: 180, CaptureMS: 60, SendMS: 90, CommitMS: 25}) { t.Errorf("move to %s within %s answered %+v", gotAddr, gotBudget, body) } for _, tt := range []struct {