diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index c556668..9a72137 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -66,6 +66,22 @@ type Config struct { PortRange [2]int // MgmtInterface is the ethernet interface used for NCCL bootstrap. MgmtInterface string + // SettleFor is how long after start to leave placement alone while agents + // reconnect. + // + // A control plane that has just started knows nothing, and the first + // reconcile tick runs before any agent has registered. It sees an empty + // fleet and acts on it: "needs 2 nodes on one fabric, but no node reports a + // fabric" is that, and it is harmless only because adoption catches up. + // What is not harmless is placing on the fraction of the fleet that + // happens to have reported first — after one restart both replicas of a + // model landed on the single node that had registered, when four were + // coming. + // + // Adoption still runs throughout; only placement waits. The window ends + // early once the fleet stops growing, so a small cluster does not sit idle + // for the whole duration. + SettleFor time.Duration } // DefaultConfig returns sensible defaults. @@ -76,6 +92,7 @@ func DefaultConfig() Config { MaxBackoff: 10 * time.Minute, UnreachableAfter: 15 * time.Second, LostAfter: 60 * time.Second, + SettleFor: 20 * time.Second, DrainGrace: 5 * time.Minute, PortRange: [2]int{20000, 20999}, MgmtInterface: "eth0", @@ -93,6 +110,11 @@ func (c *Config) normalize() { if c.MaxBackoff < c.BaseBackoff { c.MaxBackoff = max(c.BaseBackoff, d.MaxBackoff) } + if c.SettleFor < 0 { + c.SettleFor = 0 + } else if c.SettleFor == 0 { + c.SettleFor = d.SettleFor + } if c.UnreachableAfter <= 0 { c.UnreachableAfter = d.UnreachableAfter } @@ -138,6 +160,12 @@ type Reconciler struct { stop chan struct{} done chan struct{} once sync.Once + + // startedAt and lastNodeCount track the settling window: placement waits + // until the fleet has stopped growing, or the window expires. + startedAt time.Time + lastNodeCount int + steadySince time.Time } // New builds a reconciler. @@ -158,11 +186,15 @@ func New(cfg Config, cat *catalog.Catalog, inv *inventory.Store, disp Dispatcher wake: make(chan struct{}, 1), stop: make(chan struct{}), done: make(chan struct{}), + startedAt: time.Now(), } } // SetClock replaces the time source, for tests. -func (r *Reconciler) SetClock(now func() time.Time) { r.now = now } +func (r *Reconciler) SetClock(now func() time.Time) { + r.now = now + r.startedAt = now() +} // Start runs the loop until Close. func (r *Reconciler) Start(ctx context.Context) { @@ -212,11 +244,51 @@ func (r *Reconciler) Reconcile(ctx context.Context) { r.drainCordonedNodes(ctx) r.reapFailed(ctx) + // Everything above is about what is already running and stays correct on a + // partial fleet. Placement is the part that must not act on one. + if !r.fleetHasSettled() { + return + } + for _, m := range r.cat.Models() { r.reconcileModel(ctx, m) } } +// fleetHasSettled reports whether placement may run yet. +// +// Two ways out, because either alone is wrong. Waiting for the window to expire +// would make every restart cost that much downtime even when the whole fleet +// reconnected in a second. Waiting only for the count to stop changing would +// hang forever on a cluster whose nodes never all come back. +func (r *Reconciler) fleetHasSettled() bool { + if r.cfg.SettleFor <= 0 { + return true + } + now := r.now() + if now.Sub(r.startedAt) >= r.cfg.SettleFor { + return true + } + + n := len(r.inv.Nodes()) + if n != r.lastNodeCount { + r.lastNodeCount, r.steadySince = n, now + return false + } + // Steady for a moment, and at least one node present. A count that has been + // zero since start is not a settled fleet, it is one that has not arrived. + if n == 0 { + return false + } + settled := now.Sub(r.steadySince) >= r.cfg.SettleFor/4 + if settled { + r.log.Info("fleet settled; placement resuming", + "nodes", n, "waited", now.Sub(r.startedAt).Round(time.Millisecond)) + r.startedAt = time.Time{} // never wait again in this process + } + return settled +} + // expireStuckReplicas fails replicas that never finished coming up. // // Every other path out of Starting depends on the agent reporting something. A diff --git a/internal/control/reconciler/reconciler_test.go b/internal/control/reconciler/reconciler_test.go index b3cbec1..ffc4e19 100644 --- a/internal/control/reconciler/reconciler_test.go +++ b/internal/control/reconciler/reconciler_test.go @@ -141,6 +141,14 @@ func newHarness(t *testing.T, m catalog.Model, nodes ...inventory.Node) *harness for _, n := range nodes { touchAt(inv, n.ID, c.now()) } + // Past the startup settling window. These nodes are all here — the harness + // built the whole fleet before anything ran — which is the state settling + // waits for, not one it should be made to simulate in every test. The + // window's own behaviour is tested separately. + c.advance(cfg.SettleFor) + for _, n := range nodes { + touchAt(inv, n.ID, c.now()) + } return &harness{r: r, inv: inv, disp: disp, clock: c} } @@ -992,3 +1000,117 @@ func TestAHalfThatNeverArrivesStillFails(t *testing.T) { t.Errorf("node has %d free GPUs; the abandoned half still holds them", free) } } + +// --- the startup settling window --- + +func settlingHarness(t *testing.T, m catalog.Model) (*Reconciler, *inventory.Store, *fakeDispatcher, *clock) { + t.Helper() + cat, err := catalog.New([]catalog.Model{m}) + if err != nil { + t.Fatal(err) + } + inv := inventory.NewStore() + disp := &fakeDispatcher{} + c := newClock() + r := New(DefaultConfig(), cat, inv, disp, slog.New(slog.DiscardHandler)) + r.SetClock(c.now) + return r, inv, disp, c +} + +func addNode(inv *inventory.Store, c *clock, n inventory.Node) { + inv.Upsert(n) + touchAt(inv, n.ID, c.now()) +} + +func TestPlacementWaitsForTheFleetToArrive(t *testing.T) { + // A control plane that has just started knows nothing, and the first tick + // runs before any agent has registered. Placing then puts everything on + // whichever fraction reported first: after one restart both replicas of a + // model landed on the single node that had registered, when four were + // coming. + r, inv, disp, c := settlingHarness(t, testModel(2)) + + // Nothing registered yet. + r.Reconcile(t.Context()) + if got := len(disp.startedShards()); got != 0 { + t.Fatalf("placed %d shards on an empty fleet", got) + } + + // One node of four. + addNode(inv, c, gpuNode("1", 8)) + r.Reconcile(t.Context()) + if got := len(disp.startedShards()); got != 0 { + t.Fatalf("placed %d shards while the fleet was still arriving", got) + } + + // The rest arrive. + for _, id := range []string{"2", "3", "4"} { + addNode(inv, c, gpuNode(id, 8)) + } + r.Reconcile(t.Context()) + if got := len(disp.startedShards()); got != 0 { + t.Fatalf("placed %d shards the instant the last node appeared", got) + } + + // Steady for a moment: now it may act, and on the whole fleet. + c.advance(DefaultConfig().SettleFor / 4) + for _, id := range []string{"1", "2", "3", "4"} { + touchAt(inv, id, c.now()) + } + r.Reconcile(t.Context()) + if got := len(disp.startedShards()); got == 0 { + t.Fatal("placement never resumed after the fleet settled") + } +} + +func TestPlacementDoesNotWaitForeverForNodesThatNeverCome(t *testing.T) { + // Waiting only for the count to stop changing would hang forever on a + // cluster whose nodes never all come back. + r, inv, disp, c := settlingHarness(t, testModel(1)) + addNode(inv, c, gpuNode("1", 8)) + + c.advance(DefaultConfig().SettleFor) + touchAt(inv, "1", c.now()) + r.Reconcile(t.Context()) + + if got := len(disp.startedShards()); got == 0 { + t.Error("placement never resumed once the window expired") + } +} + +func TestSettlingIsOnlyEverPaidOnce(t *testing.T) { + // A node joining later in the day is an ordinary event, not a reason to + // stop placing while the cluster is running. + r, inv, disp, c := settlingHarness(t, testModel(1)) + addNode(inv, c, gpuNode("1", 8)) + c.advance(DefaultConfig().SettleFor) + touchAt(inv, "1", c.now()) + r.Reconcile(t.Context()) + if len(disp.startedShards()) == 0 { + t.Fatal("nothing placed after the window") + } + + disp.reset() + addNode(inv, c, gpuNode("2", 8)) + if !r.fleetHasSettled() { + t.Error("a node joining a running cluster restarted the settling window") + } +} + +func TestAdoptionIsNotBlockedByTheSettlingWindow(t *testing.T) { + // Only placement waits. Adoption is how a restart recovers what is already + // running, and delaying it would leave live engines unaccounted for exactly + // when the reconciler is deciding what to build. + r, inv, _, c := settlingHarness(t, testModel(1)) + addNode(inv, c, gpuNode("1", 8)) + + unclaimed := r.Adopt("1", []AdoptedShard{ + adopted("llama-99", "llama-99-r0", "1", 0, []int{0, 1, 2, 3}), + }) + if len(unclaimed) != 0 { + t.Fatalf("adoption was refused during settling: %+v", unclaimed) + } + if got := len(r.Replicas()); got != 1 { + t.Fatalf("expected the replica to be adopted, got %d", got) + } +}