diff --git a/internal/saas/controlplane/controlplane.go b/internal/saas/controlplane/controlplane.go index 0f62aec9..ff30500c 100644 --- a/internal/saas/controlplane/controlplane.go +++ b/internal/saas/controlplane/controlplane.go @@ -22,6 +22,7 @@ package controlplane import ( "net/http" + "sync" "time" "sigs.k8s.io/controller-runtime/pkg/client" @@ -56,6 +57,13 @@ type K8sControlPlane struct { // checks still apply. singleTenantNamespace string now func() time.Time + + // poolSeen caches POSITIVE pool-existence pre-check results per + // namespace/name until the stored expiry, so a repeat create of a stable + // hosted pool skips the serialized apiserver Get (see poolCheckTTL in + // forward.go). Absence is never stored. + poolSeenMu sync.Mutex + poolSeen map[string]time.Time } // Option configures a K8sControlPlane. diff --git a/internal/saas/controlplane/forward.go b/internal/saas/controlplane/forward.go index 907d5572..f9c86f45 100644 --- a/internal/saas/controlplane/forward.go +++ b/internal/saas/controlplane/forward.go @@ -15,6 +15,7 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/watch" "sigs.k8s.io/controller-runtime/pkg/client" v1 "mitos.run/mitos/api/v1" @@ -28,6 +29,13 @@ import ( // token and endpoint. const tokenSecretSuffix = "-sandbox-token" +// poolCheckTTL bounds the positive pool-existence cache feeding create's typo +// fast-fail pre-check. Hosted pools are pre-provisioned and stable, so a pool +// seen once is not re-read on every create; a pool deleted inside the window +// falls through to the controller's bounded grace (#637), the same path a +// direct create takes. Absence is NEVER cached. +const poolCheckTTL = 30 * time.Second + // maxCreateReplicas is the inclusive upper bound on the replicas a single create // request may ask for. It is a request-validation guard against a huge value // churning the controller; real fleet size is separately bounded by controller @@ -184,12 +192,22 @@ func (k *K8sControlPlane) create(ctx context.Context, req saas.ForwardRequest) ( // would resolve the poolRef in. A transient (non-NotFound) read error is NOT // treated as absent: the create proceeds and the controller's bounded grace // still governs the direct and GitOps-race paths. - var poolObj v1.SandboxPool - if err := k.c.Get(ctx, client.ObjectKey{Namespace: ns, Name: pool}, &poolObj); err != nil && apierrors.IsNotFound(err) { - return errResp(apierr.Get(apierr.CodeNotFound). - WithMessage(fmt.Sprintf("no such pool %q", pool)). - WithCause(fmt.Sprintf("no SandboxPool named %q exists in namespace %q", pool, ns)). - WithRemediation("Check the pool name for a typo and use a pool listed by GET /v1/templates, or create the SandboxPool before launching a sandbox from it.")), nil + // A POSITIVE result from a previous create inside poolCheckTTL skips the + // round trip: hosted pools are stable, and the pre-check is a typo UX + // guard, not a correctness gate (a pool deleted inside the window falls + // through to the controller's bounded grace, #637). Absence is never + // cached, so a genuine typo re-reads authoritatively every time. + if !k.poolRecentlySeen(ns, pool) { + var poolObj v1.SandboxPool + switch err := k.c.Get(ctx, client.ObjectKey{Namespace: ns, Name: pool}, &poolObj); { + case err != nil && apierrors.IsNotFound(err): + return errResp(apierr.Get(apierr.CodeNotFound). + WithMessage(fmt.Sprintf("no such pool %q", pool)). + WithCause(fmt.Sprintf("no SandboxPool named %q exists in namespace %q", pool, ns)). + WithRemediation("Check the pool name for a typo and use a pool listed by GET /v1/templates, or create the SandboxPool before launching a sandbox from it.")), nil + case err == nil: + k.markPoolSeen(ns, pool) + } } name := generateName() @@ -233,6 +251,26 @@ func (k *K8sControlPlane) create(ctx context.Context, req saas.ForwardRequest) ( sb.Spec.Lifetime = &v1.SandboxLifetime{TTL: &metav1.Duration{Duration: d}} } + // Establish the ready watch BEFORE the Create: the create event then + // arrives on the stream itself, which both closes the flip-before-watch + // window (no authoritative re-read needed) and takes one serialized + // apiserver round trip off the create hot path. On any establish failure + // the create proceeds and the wait falls back to the post-create watch or + // poll (pollReady), exactly as before. + var wi watch.Interface + if w, ok := k.c.(client.WithWatch); ok { + // A dedicated cancel bounds the watch stream's lifetime to this create. + watchCtx, cancel := context.WithCancel(ctx) + defer cancel() + if established, werr := establishSandboxWatch(watchCtx, w, ns, name); werr == nil { + wi = established + defer wi.Stop() + } else { + slog.Warn("could not establish the sandbox ready watch before create; will wait post-create", + "namespace", ns, "sandbox", name, "error", werr.Error()) + } + } + if err := k.c.Create(ctx, sb); err != nil { switch { case apierrors.IsAlreadyExists(err): @@ -253,9 +291,47 @@ func (k *K8sControlPlane) create(ctx context.Context, req saas.ForwardRequest) ( } } + if wi != nil { + // The deadline starts where pollReady's would: after the Create. + deadline := k.now().Add(k.readyTimeout) + if resp, done := k.watchWait(ctx, wi, ns, name, startedAt, deadline, "Pending"); done { + return resp, nil + } + slog.Warn("sandbox ready watch closed before a terminal outcome; falling back to status polling", + "namespace", ns, "sandbox", name) + return k.pollReadyTicker(ctx, ns, name, startedAt, deadline) + } return k.pollReady(ctx, ns, name, startedAt) } +// poolRecentlySeen reports whether the ns/pool existence pre-check returned +// POSITIVE within poolCheckTTL. Expired entries are deleted on read so the map +// stays bounded by the set of live pools. +func (k *K8sControlPlane) poolRecentlySeen(ns, pool string) bool { + key := ns + "/" + pool + k.poolSeenMu.Lock() + defer k.poolSeenMu.Unlock() + exp, ok := k.poolSeen[key] + if !ok { + return false + } + if k.now().After(exp) { + delete(k.poolSeen, key) + return false + } + return true +} + +// markPoolSeen records a POSITIVE ns/pool existence result until now+poolCheckTTL. +func (k *K8sControlPlane) markPoolSeen(ns, pool string) { + k.poolSeenMu.Lock() + defer k.poolSeenMu.Unlock() + if k.poolSeen == nil { + k.poolSeen = make(map[string]time.Time) + } + k.poolSeen[ns+"/"+pool] = k.now().Add(poolCheckTTL) +} + // pollReady blocks until the sandbox reaches Ready (returns 201 + token), Failed // (returns the rejection message), a terminal Rejected condition (409 with the // controller's actionable message; the fork engine records it without a Failed diff --git a/internal/saas/controlplane/poolcheck_test.go b/internal/saas/controlplane/poolcheck_test.go new file mode 100644 index 00000000..17715d21 --- /dev/null +++ b/internal/saas/controlplane/poolcheck_test.go @@ -0,0 +1,144 @@ +package controlplane + +import ( + "context" + "net/http" + "strings" + "sync/atomic" + "testing" + "time" + + "sigs.k8s.io/controller-runtime/pkg/client" + fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + v1 "mitos.run/mitos/api/v1" + "mitos.run/mitos/internal/saas" +) + +// poolGetCountingClient wraps the fake client and counts Gets of SandboxPool +// objects, the typo fast-fail pre-check round trip on the create hot path. +func poolGetCountingClient(t *testing.T, gets *atomic.Int64, objs ...client.Object) client.Client { + t.Helper() + base := fakeclient.NewClientBuilder(). + WithScheme(newScheme(t)). + WithStatusSubresource(&v1.Sandbox{}). + WithObjects(objs...). + Build() + return interceptor.NewClient(base, interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*v1.SandboxPool); ok { + gets.Add(1) + } + return cl.Get(ctx, key, obj, opts...) + }, + }) +} + +// createOnce drives one create through Forward as org, flipping the sandbox +// Ready in the background, and returns the response. +func createOnce(t *testing.T, cp *K8sControlPlane, c client.Client, org, pool string) saas.ForwardResponse { + t.Helper() + stop := flipToReadyWhenCreated(t, c, org, "10.1.2.3:9091", "tok-pool") + defer stop() + resp, err := cp.Forward(context.Background(), saas.ForwardRequest{ + OrgID: org, Op: "sandbox.create", Body: []byte(`{"pool":"` + pool + `"}`), + }) + if err != nil { + t.Fatalf("Forward: %v", err) + } + return resp +} + +// TestRepeatCreateSkipsThePoolPreCheckRoundTrip asserts the typo fast-fail +// pool pre-check is paid at most once per TTL window: hosted pools are +// pre-provisioned and stable, so a positive existence result is cached and a +// repeat create of the same pool performs no SandboxPool Get. The pre-check is +// a UX guard, not a correctness gate; a pool deleted inside the window falls +// through to the controller's bounded grace (#637), the same path a direct +// create takes. +func TestRepeatCreateSkipsThePoolPreCheckRoundTrip(t *testing.T) { + var poolGets atomic.Int64 + c := poolGetCountingClient(t, &poolGets, poolIn(orgA, "default")) + cp := New(c, WithPollInterval(5*time.Millisecond), WithReadyTimeout(2*time.Second)) + + if resp := createOnce(t, cp, c, orgA, "default"); resp.Status != http.StatusCreated { + t.Fatalf("first create status = %d, body = %s", resp.Status, resp.Body) + } + if resp := createOnce(t, cp, c, orgA, "default"); resp.Status != http.StatusCreated { + t.Fatalf("second create status = %d, body = %s", resp.Status, resp.Body) + } + if n := poolGets.Load(); n != 1 { + t.Fatalf("two creates of one stable pool performed %d SandboxPool Get(s), want 1: the positive pre-check must be cached", n) + } +} + +// TestPoolPreCheckCacheIsNamespaceScoped asserts a positive cache entry for +// one org's namespace never satisfies the pre-check for the SAME pool name in +// another org's namespace: the cache key carries the namespace, so each tenant +// pays (and proves) its own existence check. +func TestPoolPreCheckCacheIsNamespaceScoped(t *testing.T) { + var poolGets atomic.Int64 + c := poolGetCountingClient(t, &poolGets, poolIn(orgA, "default"), poolIn(orgB, "default")) + cp := New(c, WithPollInterval(5*time.Millisecond), WithReadyTimeout(2*time.Second)) + + if resp := createOnce(t, cp, c, orgA, "default"); resp.Status != http.StatusCreated { + t.Fatalf("org A create status = %d, body = %s", resp.Status, resp.Body) + } + if resp := createOnce(t, cp, c, orgB, "default"); resp.Status != http.StatusCreated { + t.Fatalf("org B create status = %d, body = %s", resp.Status, resp.Body) + } + if n := poolGets.Load(); n != 2 { + t.Fatalf("two orgs creating the same pool name performed %d SandboxPool Get(s), want 2: a cache hit must never cross namespaces", n) + } +} + +// TestUnknownPoolIsReCheckedEveryCreate asserts absence is NEVER cached: every +// create naming a missing pool re-reads authoritatively and 404s instantly, so +// a pool that appears between two attempts is seen immediately. +func TestUnknownPoolIsReCheckedEveryCreate(t *testing.T) { + var poolGets atomic.Int64 + c := poolGetCountingClient(t, &poolGets) + cp := New(c, WithPollInterval(5*time.Millisecond), WithReadyTimeout(2*time.Second)) + + for i := 0; i < 2; i++ { + resp, err := cp.Forward(context.Background(), saas.ForwardRequest{ + OrgID: orgA, Op: "sandbox.create", Body: []byte(`{"pool":"nope"}`), + }) + if err != nil { + t.Fatalf("Forward: %v", err) + } + if resp.Status != http.StatusNotFound { + t.Fatalf("create %d status = %d, want 404; body = %s", i, resp.Status, resp.Body) + } + if !strings.Contains(string(resp.Body), "no such pool") || !strings.Contains(string(resp.Body), "nope") { + t.Errorf("create %d body missing the pool name: %s", i, resp.Body) + } + } + if n := poolGets.Load(); n != 2 { + t.Fatalf("two creates of a missing pool performed %d SandboxPool Get(s), want 2: absence must never be cached", n) + } +} + +// TestPoolPreCheckCacheExpires asserts the positive cache is bounded by its +// TTL: once the window passes, the next create re-reads authoritatively. +func TestPoolPreCheckCacheExpires(t *testing.T) { + var poolGets atomic.Int64 + c := poolGetCountingClient(t, &poolGets, poolIn(orgA, "default")) + cp := New(c, WithPollInterval(5*time.Millisecond), WithReadyTimeout(2*time.Second)) + + if resp := createOnce(t, cp, c, orgA, "default"); resp.Status != http.StatusCreated { + t.Fatalf("first create status = %d, body = %s", resp.Status, resp.Body) + } + + // Step the control plane's clock past the TTL; timers derived from it stay + // sane because the offset is constant. + cp.now = func() time.Time { return time.Now().Add(poolCheckTTL + time.Second) } + + if resp := createOnce(t, cp, c, orgA, "default"); resp.Status != http.StatusCreated { + t.Fatalf("second create status = %d, body = %s", resp.Status, resp.Body) + } + if n := poolGets.Load(); n != 2 { + t.Fatalf("a create after the TTL performed %d total SandboxPool Get(s), want 2: the cache must expire", n) + } +} diff --git a/internal/saas/controlplane/readywatch.go b/internal/saas/controlplane/readywatch.go index 3c36fa06..a1a09d8b 100644 --- a/internal/saas/controlplane/readywatch.go +++ b/internal/saas/controlplane/readywatch.go @@ -33,14 +33,7 @@ func (k *K8sControlPlane) watchReady(ctx context.Context, w client.WithWatch, ns watchCtx, cancel := context.WithCancel(ctx) defer cancel() - var list v1.SandboxList - wi, err := w.Watch(watchCtx, &list, - client.InNamespace(ns), - // The api server filters to the single object server-side. The fake - // client ignores field selectors on Watch, so the loop below re-checks - // the name on every event regardless. - client.MatchingFieldsSelector{Selector: fields.OneTermEqualSelector("metadata.name", name)}, - ) + wi, err := establishSandboxWatch(watchCtx, w, ns, name) if err != nil { slog.Warn("could not establish the sandbox ready watch", "namespace", ns, "sandbox", name, "error", err.Error()) @@ -50,7 +43,9 @@ func (k *K8sControlPlane) watchReady(ctx context.Context, w client.WithWatch, ns // Authoritative read AFTER the watch is established: a phase flip between // the Create and the Watch would otherwise never produce an event and the - // wait would idle until the fallback re-Get. + // wait would idle until the fallback re-Get. (The create hot path avoids + // this read entirely by establishing its watch BEFORE the Create; this + // path serves waits whose object predates the watch, such as fork.) var sb v1.Sandbox if err := k.c.Get(ctx, client.ObjectKey{Namespace: ns, Name: name}, &sb); err != nil { return readSandboxError(ctx, err), true @@ -58,8 +53,32 @@ func (k *K8sControlPlane) watchReady(ctx context.Context, w client.WithWatch, ns if resp, done := k.sandboxOutcome(ctx, &sb, startedAt); done { return resp, true } - lastPhase := phaseOrUnknown(&sb) + return k.watchWait(ctx, wi, ns, name, startedAt, deadline, phaseOrUnknown(&sb)) +} + +// establishSandboxWatch opens the single-object watch backing the readiness +// waits. The create hot path calls it BEFORE the Sandbox Create, so the create +// event itself arrives on the stream and no authoritative re-read is needed to +// close a missed-event window; watchReady calls it after, for objects that +// already exist. +func establishSandboxWatch(ctx context.Context, w client.WithWatch, ns, name string) (watch.Interface, error) { + var list v1.SandboxList + return w.Watch(ctx, &list, + client.InNamespace(ns), + // The api server filters to the single object server-side. The fake + // client ignores field selectors on Watch, so the event loop re-checks + // the name on every event regardless. Watching a name that does not + // exist yet is valid; events begin when the object is created. + client.MatchingFieldsSelector{Selector: fields.OneTermEqualSelector("metadata.name", name)}, + ) +} + +// watchWait consumes an ALREADY-ESTABLISHED single-object watch until a +// terminal create outcome, the deadline, or a closed stream (done=false: the +// caller fails open to polling for the remaining deadline budget). lastPhase +// seeds the phase a timeout envelope names before any event arrives. +func (k *K8sControlPlane) watchWait(ctx context.Context, wi watch.Interface, ns, name string, startedAt time.Time, deadline time.Time, lastPhase string) (saas.ForwardResponse, bool) { // The overall ready timeout: a non-positive remainder fires immediately. timeout := time.NewTimer(deadline.Sub(k.now())) defer timeout.Stop() diff --git a/internal/saas/controlplane/readywatch_test.go b/internal/saas/controlplane/readywatch_test.go index 9a21b394..465277e4 100644 --- a/internal/saas/controlplane/readywatch_test.go +++ b/internal/saas/controlplane/readywatch_test.go @@ -5,9 +5,11 @@ import ( "errors" "net/http" "strings" + "sync/atomic" "testing" "time" + corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/watch" "sigs.k8s.io/controller-runtime/pkg/client" @@ -122,6 +124,99 @@ func TestWatchDeletedMidCreateIsNotFound(t *testing.T) { } } +// TestWatchDrivenCreateNeedsNoSandboxGet asserts the watch-driven create hot +// path performs ZERO Gets on the Sandbox object: the ready watch is established +// BEFORE the Create, so the create event itself arrives on the watch and no +// authoritative re-read is needed to close a missed-event window. Every +// serialized apiserver round trip here is paid by every hosted create (prod +// measured the control-plane share of create at ~140 ms P50 against a 60-76 ms +// engine restore), so the round-trip count is the property under test, the same +// way the SDK pins its connection count. +func TestWatchDrivenCreateNeedsNoSandboxGet(t *testing.T) { + var sandboxGets atomic.Int64 + base := fakeclient.NewClientBuilder(). + WithScheme(newScheme(t)). + WithStatusSubresource(&v1.Sandbox{}). + WithObjects(poolIn(orgA, "default")). + Build() + c := interceptor.NewClient(base, interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*v1.Sandbox); ok { + sandboxGets.Add(1) + } + return cl.Get(ctx, key, obj, opts...) + }, + }) + cp := New(c, WithPollInterval(3*time.Second), WithReadyTimeout(10*time.Second)) + + stop := flipToReadyWhenCreated(t, c, orgA, "10.1.2.3:9091", "tok-noget") + defer stop() + + resp, err := cp.Forward(context.Background(), saas.ForwardRequest{ + OrgID: orgA, Op: "sandbox.create", Body: []byte(`{"pool":"default"}`), + }) + if err != nil { + t.Fatalf("Forward: %v", err) + } + if resp.Status != http.StatusCreated { + t.Fatalf("status = %d, body = %s", resp.Status, resp.Body) + } + if n := sandboxGets.Load(); n != 0 { + t.Fatalf("watch-driven create performed %d Sandbox Get(s); the watch must be established before the Create so no authoritative re-read is needed", n) + } +} + +// TestReadyFlipDuringCreateWriteIsStillSeen asserts a phase flip that lands +// BEFORE the Create call even returns (the fastest controller imaginable) is +// still observed. This is the race the post-establish authoritative Get used +// to close; with the watch established before the Create, the events +// themselves cover it, and this test keeps that property pinned. +func TestReadyFlipDuringCreateWriteIsStillSeen(t *testing.T) { + base := fakeclient.NewClientBuilder(). + WithScheme(newScheme(t)). + WithStatusSubresource(&v1.Sandbox{}). + WithObjects(poolIn(orgA, "default")). + Build() + c := interceptor.NewClient(base, interceptor.Funcs{ + Create: func(ctx context.Context, cl client.WithWatch, obj client.Object, opts ...client.CreateOption) error { + if err := cl.Create(ctx, obj, opts...); err != nil { + return err + } + sb, ok := obj.(*v1.Sandbox) + if !ok { + return nil + } + ready := sb.DeepCopy() + ready.Status.Phase = v1.SandboxReady + ready.Status.Endpoint = "10.9.9.9:9091" + if err := cl.Status().Update(ctx, ready); err != nil { + return err + } + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: sb.Name + tokenSecretSuffix, Namespace: sb.Namespace}, + Data: map[string][]byte{"token": []byte("tok-flip"), "endpoint": []byte("10.9.9.9:9091")}, + } + return cl.Create(ctx, secret) + }, + }) + cp := New(c, WithPollInterval(3*time.Second), WithReadyTimeout(10*time.Second)) + + started := time.Now() + resp, err := cp.Forward(context.Background(), saas.ForwardRequest{ + OrgID: orgA, Op: "sandbox.create", Body: []byte(`{"pool":"default"}`), + }) + elapsed := time.Since(started) + if err != nil { + t.Fatalf("Forward: %v", err) + } + if resp.Status != http.StatusCreated { + t.Fatalf("status = %d, body = %s", resp.Status, resp.Body) + } + if elapsed >= 1*time.Second { + t.Fatalf("Ready flip during the Create write surfaced after %s; it must be seen immediately, not on a tick or fallback boundary", elapsed) + } +} + // TestWatchEstablishFailureFallsBackToPolling asserts the create fails OPEN to // the legacy poll loop when the watch cannot be established: readiness is // still observed and the create still succeeds.