diff --git a/cmd/main.go b/cmd/main.go index 48908e6..dcf2859 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -64,13 +64,20 @@ import ( // a fallback for managerNamespace when POD_NAMESPACE is unset. const defaultNamespace = "nebula-system" -// restConfigQPS and restConfigBurst size the client-go bucket shared by every API call in the -// process — each controller's, and the virtual kubelet's status pushes. controller-runtime's -// default 20/30 is what binds first at fleet scale, and it binds invisibly: throttled calls -// wait in our own process, so it reads as API-server or provider slowness. +// restConfigQPS and restConfigBurst size the bucket client-go installs for EACH clientset built +// from this config — the manager's, the virtual kubelet's, the cert bootstrapper's two — and NOT +// one process-wide budget: NewForConfigAndClient only creates a limiter when RateLimiter is nil, +// on its own shallow copy. Left unshared for the reason vnode.podQueueRateLimiter gives: one +// budget lets status pushes starve the reconcilers into a stall nobody can read. +// +// 150 is sized for 1k workloads on the MANAGER's bucket, which is the one every controller and +// the cost-accrual loop draw on. Accrual is its floor: one write per billing claim per +// accrualInterval is ~33/s sustained, which 50 could not carry alongside placement and +// provisioning. controller-runtime's default 20/30 binds even sooner, and invisibly — throttled +// calls wait in our own process, so it reads as API-server or provider slowness. const ( - restConfigQPS = 50 - restConfigBurst = 100 + restConfigQPS = 150 + restConfigBurst = 300 ) var ( diff --git a/internal/controller/concurrency.go b/internal/controller/concurrency.go index 597d1b1..9a44d3f 100644 --- a/internal/controller/concurrency.go +++ b/internal/controller/concurrency.go @@ -26,11 +26,11 @@ package controller // trip, or a provider call — not by CPU. So the worker sits idle while the fleet backs up, // and the observed rate is 1/latency regardless of how much CPU the manager is given. // -// 8 matches pkg/vnode's podSyncWorkers, deliberately: the two pipelines hand work to each -// other (VK writes the Pod status the claim controller waits on, and the claim controller's -// teardown follows VK's DeletePod), so sizing them alike keeps either from being the -// other's ceiling. Raising it further trades API-server pressure for latency, and the -// server, not this constant, is the next limit. +// Deliberately NOT matched to pkg/vnode's podSyncWorkers, which is far higher: the two are +// bounded by different things. A VK worker blocks on a provider call and holds no API token, +// so width there is nearly free, while every reconcile here is API writes — this one is +// bounded by the client's rate budget (see restConfigQPS in cmd/main.go), and the server, not +// this constant, is the next limit. // // Safe because controller-runtime never reconciles the same key concurrently, so per-object // state needs no locking, and the only state shared ACROSS objects is read-only (the diff --git a/internal/controller/cost_accrual.go b/internal/controller/cost_accrual.go index cc4aff8..1158812 100644 --- a/internal/controller/cost_accrual.go +++ b/internal/controller/cost_accrual.go @@ -20,6 +20,7 @@ import ( "context" "math" "strconv" + "sync" "time" corev1 "k8s.io/api/core/v1" @@ -41,8 +42,8 @@ import ( // What it does bound is the gap a baseline has to survive — a scrape must land between a series' // zero sample and its first charge, one tick later, or those dollars reach no increase() query at // all (see seedClaimBaseline). Thirty seconds keeps that reachable for the usual 15s scrape, at -// ~16.7 writes/s across a 500-claim fleet: a third of the client's rate budget (see the QPS in -// cmd/main.go), which is where this stops being free and starts competing with the reconcilers. +// ~33 writes/s across the 1k fleet this targets — a fifth of the client's rate budget (see +// restConfigQPS), which is where this stops being free and competes with the reconcilers. const accrualInterval = 30 * time.Second // accrualTimeout bounds one whole tick, List plus every write. A tick that cannot finish loses @@ -54,6 +55,18 @@ const accrualInterval = 30 * time.Second // back-to-back ticks, and re-deriving a window from its anchor cannot double-charge it. const accrualTimeout = accrualInterval +// accrualWorkers is how many claims one tick checkpoints at once. +// +// A tick is one Update per billing claim, so a 1k fleet is 1k round trips — more than +// accrualTimeout allows serially. That failure is not random: List order is stable, so the claims +// at its tail are the SAME ones whose EST_COST stalls every tick. Concurrency is what fits the +// fleet inside the window; it does not change how many writes a tick makes. +// +// Bounded, because the writes still share one client rate budget (see restConfigQPS in +// cmd/main.go) — past that ceiling extra goroutines queue inside this process instead of at the +// API server, which is the harder stall to read. +const accrualWorkers = 16 + // CostAccrual advances each claim's durable spend ledger on a ticker. // // A Runnable rather than a hook on the reconcile path: spend accrues with the CLOCK, not with @@ -113,18 +126,30 @@ func (a *CostAccrual) accrueAll(ctx context.Context) { log.Error(err, "listing nodeclaims to accrue") return } + // Each claim is its own object and its own compare-and-swap, so width costs nothing in + // correctness — see accrualWorkers for what it does cost. + sem := make(chan struct{}, accrualWorkers) + var wg sync.WaitGroup for i := range claims.Items { nc := &claims.Items[i] - if err := a.accrue(ctx, nc); err != nil { - // A conflict is the ordinary case — the reconciler patched the same claim from its - // own copy. The anchor did not move, so this window is simply charged next tick. - if apierrors.IsConflict(err) || apierrors.IsNotFound(err) { - log.V(1).Info("skipping accrual this tick", "claim", nc.Name, "reason", err.Error()) - continue + sem <- struct{}{} + wg.Add(1) + go func() { + defer wg.Done() + defer func() { <-sem }() + if err := a.accrue(ctx, nc); err != nil { + // A conflict is the ordinary case — the reconciler patched the same claim from + // its own copy. The anchor did not move, so this window is simply charged next + // tick. + if apierrors.IsConflict(err) || apierrors.IsNotFound(err) { + log.V(1).Info("skipping accrual this tick", "claim", nc.Name, "reason", err.Error()) + return + } + log.Error(err, "accruing claim cost", "claim", nc.Name) } - log.Error(err, "accruing claim cost", "claim", nc.Name) - } + }() } + wg.Wait() } // accrue persists what the claim has cost as of now and re-anchors there, in one patch. diff --git a/internal/controller/cost_accrual_test.go b/internal/controller/cost_accrual_test.go index bd15971..108b827 100644 --- a/internal/controller/cost_accrual_test.go +++ b/internal/controller/cost_accrual_test.go @@ -19,6 +19,7 @@ package controller import ( "context" "errors" + "fmt" "math" "reflect" "strconv" @@ -139,6 +140,45 @@ func TestCostAccrual_ChargesFromTheAnchor(t *testing.T) { // The whole point of persisting a timestamp: a window that spans a restart is charged in full on // the first tick back, not clipped to one interval. +// A fleet wider than accrualWorkers, which is the only shape that exercises the fan-out at all: a +// single-claim tick runs one goroutine and proves nothing. Charged exactly once each is the whole +// property — a lost claim and a doubly-charged one both read as a wrong counter. +func TestCostAccrual_ChargesEveryClaimOfAWideFleet(t *testing.T) { + const fleetSize = accrualWorkers * 4 + halfHour := 30 * time.Minute + // One pinned instant for every anchor AND for the accrual clock. billingClaim reads + // time.Now() per claim, so a second boundary crossing anywhere in the fan-out charges the + // claims on one side of it a 1s-longer window — $0.027 at this rate, far outside the + // tolerances below. + pinned := time.Now().Truncate(time.Second) + start := metav1.Time{Time: pinned.Add(-halfHour)} + fleet := make([]*nebulav1alpha1.NodeClaim, 0, fleetSize) + for i := range fleetSize { + nc := billingClaim(fmt.Sprintf("bound-%d", i), "98.3200", &halfHour) + nc.Status.ProvisionedAt = start.DeepCopy() + nc.Status.LastAccruedAt = start.DeepCopy() + fleet = append(fleet, nc) + } + a, c := newAccrual(t, fleet...) + a.now = func() time.Time { return pinned } + + a.accrueAll(context.Background()) + + want := 98.32 * 0.5 + for _, nc := range fleet { + total, anchor := ledger(t, c, nc.Name) + if math.Abs(total-want) > 1e-3 { + t.Fatalf("claim %q: status.estimatedCostUSD %v, want %v", nc.Name, total, want) + } + if anchor == nil || time.Since(anchor.Time) > time.Minute { + t.Fatalf("claim %q: anchor was not moved forward: %v", nc.Name, anchor) + } + } + if got, wantAll := booked(t), want*fleetSize; math.Abs(got-wantAll) > 1e-6 { + t.Fatalf("booked %v, want %v — a claim was charged twice or not at all", got, wantAll) + } +} + func TestCostAccrual_RecoversDowntime(t *testing.T) { down := 3 * time.Hour a, c := newAccrual(t, billingClaim("bound", "10.0000", &down)) diff --git a/pkg/vnode/node.go b/pkg/vnode/node.go index a44aa1c..b537f71 100644 --- a/pkg/vnode/node.go +++ b/pkg/vnode/node.go @@ -47,7 +47,7 @@ const informerResync = time.Minute // work per pod key, so distinct pods provision in parallel while one key never runs // twice — without this, a single slow provision blocks pods that would succeed // instantly. -const podSyncWorkers = 32 +const podSyncWorkers = 64 // podQueueRate and podQueueBurst size the token bucket that admits work into each of the // pod controller's queues.