diff --git a/api/v1alpha1/groupversion_info.go b/api/v1alpha1/groupversion_info.go index 93fe73e..eff97d4 100644 --- a/api/v1alpha1/groupversion_info.go +++ b/api/v1alpha1/groupversion_info.go @@ -130,6 +130,24 @@ const ( // This one flows outward — VK writes, operators read. EndpointAnnotation = "nebula.inftyai.com/endpoint" + // InstanceIDAnnotation carries the provider's id for the external instance backing + // this Pod. Written by the virtual kubelet as soon as Provision returns an id — which + // is the only place it is ever learned, since VK otherwise holds it in memory — and + // never cleared. + // + // It exists so the NodeClaim controller can record status.InstanceID from the Pod it + // has already fetched. Before this it asked the PROVIDER, listing every instance and + // matching on claim name, on every reconcile until the id resolved: correct, but a + // provider API call per reconcile per claim, which against a real backend means + // hundreds of DescribeInstances/list calls for one large batch, into APIs that rate + // limit. The id is a fact VK already knows, so it flows outward on the Pod like the + // endpoint does rather than being searched for. + // + // The claim's own copy is still the durable one: teardown runs after the Pod is gone, + // so it reads status.InstanceID, falling back to List-by-claim-name when the id never + // made it across. + InstanceIDAnnotation = "nebula.inftyai.com/instance-id" + // TerminateInstanceFinalizer is held by every NodeClaim to guarantee teardown. VK // owns the happy path (DeletePod → provider.Terminate), but its teardown is // edge-triggered and its tracking in-memory, so a Pod force-deleted during a VK diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index 2745363..fbe0e7f 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -149,15 +149,25 @@ spec: port: 8081 initialDelaySeconds: 5 periodSeconds: 10 - # TODO(user): Configure the resources accordingly based on the project requirements. + # The manager is not a leaf controller: it runs one virtual kubelet per provider, + # each with its own pod-sync workers and poll loop, plus the placement and claim + # controllers, so its cost scales with the FLEET, not with the number of CRDs. The + # kubebuilder default (500m / 128Mi) is a scaffold value — at a few hundred Pods it + # is a CFS throttle on the very loops that are supposed to keep up. + # + # Requests EQUAL limits, which is Guaranteed QoS: no CFS throttling of the sync + # loops, and the kubelet evicts this Pod last under node pressure — the manager + # going down takes provisioning and teardown with it. The cost is that the whole + # amount must be allocatable on one node for the Pod to schedule at all, so a small + # dev cluster (Kind on a laptop with a modest Docker VM) may need these lowered. # More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/ resources: limits: - cpu: 500m - memory: 128Mi + cpu: "1" + memory: 1Gi requests: - cpu: 10m - memory: 64Mi + cpu: "1" + memory: 1Gi volumeMounts: - name: catalog mountPath: /etc/nebula/catalog diff --git a/go.mod b/go.mod index 95873e5..30a446b 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/prometheus/client_golang v1.22.0 github.com/prometheus/client_model v0.6.1 github.com/virtual-kubelet/virtual-kubelet v1.11.0 + golang.org/x/time v0.9.0 google.golang.org/grpc v1.78.0 k8s.io/api v0.33.4 k8s.io/apimachinery v0.33.4 @@ -110,7 +111,6 @@ require ( golang.org/x/sys v0.41.0 // indirect golang.org/x/term v0.40.0 // indirect golang.org/x/text v0.34.0 // indirect - golang.org/x/time v0.9.0 // indirect golang.org/x/tools v0.42.0 // indirect gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda // indirect diff --git a/internal/controller/concurrency.go b/internal/controller/concurrency.go new file mode 100644 index 0000000..597d1b1 --- /dev/null +++ b/internal/controller/concurrency.go @@ -0,0 +1,43 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +// concurrentReconciles is how many objects a FLEET-SCALED controller reconciles at once — +// one whose object count tracks the number of workloads (NodeClaims, Pods, Sandboxes) +// rather than the number of policies. +// +// controller-runtime defaults to 1, which is right for a controller with a handful of +// objects and wrong for these: a single worker turns the whole fleet into a queue served +// one item at a time, and each item's cost is dominated by WAITING — an API write round +// 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. +// +// 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 +// provider registry) or already guarded (pkg/failover.Blocklist). +// +// Not applied to the NodePool or SandboxSet controllers: their object counts track policy, +// not fleet size, and their fan-in watches collapse thousands of child events onto a few +// keys (the workqueue dedups by key), so a second worker would mostly add conflict retries +// on the same status. +const concurrentReconciles = 8 diff --git a/internal/controller/nodeclaim_controller.go b/internal/controller/nodeclaim_controller.go index f48446f..884f7d7 100644 --- a/internal/controller/nodeclaim_controller.go +++ b/internal/controller/nodeclaim_controller.go @@ -26,6 +26,7 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" "sigs.k8s.io/controller-runtime/pkg/handler" logf "sigs.k8s.io/controller-runtime/pkg/log" @@ -71,8 +72,8 @@ const podReasonInitializing = nebulav1alpha1.PodReasonInitializing // the served Pod is gone it self-deletes, and its finalizer reclaims the instance (resolve // provider → find by claim name via List → Terminate) independent of VK liveness. // -// Self-delete is guarded by placementGracePeriod so cache lag never tears down a live -// workload. +// Self-delete is guarded by placementGracePeriod, but only for a claim that has never +// observed its Pod, so cache lag never tears down a live workload. type NodeClaimReconciler struct { client.Client Scheme *runtime.Scheme @@ -125,32 +126,30 @@ func (r *NodeClaimReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( if pod != nil { // The workload is present. We do not mirror the Pod's fine-grained runtime // status, but we DO reflect the coarse phase that matters to the ledger. - // desiredPhase maps the served Pod onto the claim phase (Terminated / - // Terminating / Bound / Provisioning); markPhase persists it (and the instance - // id) idempotently. An empty desiredPhase means "hold the current phase, just - // keep the id fresh" — the Bound hold (see desiredPhase). - return ctrl.Result{}, r.markPhase(ctx, &nc, r.desiredPhase(&nc, pod)) + // markPhase maps the Pod onto that phase (Terminated / Terminating / Bound / + // Provisioning, see desiredPhase) and persists it, together with the instance id + // the Pod carries, idempotently. + return ctrl.Result{}, r.markPhase(ctx, &nc, pod) } // The served Pod is absent. Decide whether this is a real teardown or a // transient cache-lag false-negative. - if r.wasBound(&nc) || nc.Status.Phase == nebulav1alpha1.NodeClaimTerminating { - // We previously observed the Pod, so its disappearance is real: this is a - // teardown (normal delete, or a force-delete during a VK outage). A - // Terminating claim counts too — it was set only from a Pod carrying a - // DeletionTimestamp, so that Pod provably existed and its disappearance is - // the delete completing, not cache lag. Delete the claim so its finalizer - // fires the backstop. + if nc.Status.Phase != "" { + // Any phase proves this controller already saw the Pod alive: markPhase is the only + // writer of status.phase and runs only from the pod != nil branch above, while + // ensureClaim creates the claim with no status. So this is a real teardown — delete + // the claim so its finalizer fires the backstop. Provisioning counts too: waiting + // on it delayed every workload deleted while its instance was still booting. log.Info("served Pod is gone after being observed placed; deleting claim to trigger teardown", - "pod", nc.Spec.PodRef.Name) + "pod", nc.Spec.PodRef.Name, "phase", nc.Status.Phase) return ctrl.Result{}, r.deleteSelf(ctx, &nc) } - // Never observed the Pod running, and the cached read says it is absent. Since - // the claim is always created after its Pod, the Pod exists at the API server; - // an "absent" this early is almost certainly the informer cache lagging, not a - // real teardown. Wait out the short grace window (a Pod watch re-enqueues us as - // soon as the cache catches up) rather than tearing down a live instance. + // No phase yet, so the claim has no evidence of its own and the cached read is all + // there is. Since the claim is always created after its Pod, the Pod exists at the API + // server; an "absent" this early is almost certainly the informer cache lagging, not a + // real teardown. Wait out the short grace window (a Pod watch re-enqueues us as soon + // as the cache catches up) rather than tearing down a live instance. if age := time.Since(nc.CreationTimestamp.Time); age < placementGracePeriod { return ctrl.Result{RequeueAfter: placementGracePeriod - age}, nil } @@ -300,20 +299,20 @@ func (r *NodeClaimReconciler) desiredPhase(nc *nebulav1alpha1.NodeClaim, pod *co } } -// markPhase persists the claim's coarse phase and best-effort records -// status.InstanceID, in one idempotent status write. An empty phase means "leave -// the phase as-is" (the Bound-flap hold, see desiredPhase); the id is still -// refreshed. The provider id is the one datum that must not be lost — the teardown -// backstop reclaims the instance by it even if VK (which otherwise holds it only -// in memory) has died — so it is captured as soon as it resolves, regardless of -// phase. A no-op (phase unchanged and id already set) writes nothing. -func (r *NodeClaimReconciler) markPhase(ctx context.Context, nc *nebulav1alpha1.NodeClaim, phase nebulav1alpha1.NodeClaimPhase) error { +// markPhase reflects the served Pod into the claim's status: the coarse phase +// desiredPhase maps it to, plus best-effort status.InstanceID, in one idempotent +// write. An empty desiredPhase means "leave the phase as-is" (the Bound-flap hold); +// the id is still refreshed. The provider id is the one datum that must not be lost — +// the teardown backstop reclaims the instance by it even if VK (which otherwise holds +// it only in memory) has died — so it is captured as soon as the Pod carries it, +// regardless of phase. A no-op (phase unchanged and id already set) writes nothing. +func (r *NodeClaimReconciler) markPhase(ctx context.Context, nc *nebulav1alpha1.NodeClaim, pod *corev1.Pod) error { changed := false - if phase != "" && nc.Status.Phase != phase { + if phase := r.desiredPhase(nc, pod); phase != "" && nc.Status.Phase != phase { nc.Status.Phase = phase changed = true } - if r.recordInstanceID(ctx, nc) { + if recordInstanceID(nc, pod) { changed = true } if !changed { @@ -322,21 +321,23 @@ func (r *NodeClaimReconciler) markPhase(ctx context.Context, nc *nebulav1alpha1. return r.patchStatus(ctx, nc) } -// recordInstanceID resolves and stores status.InstanceID when it is not already -// set, returning whether it mutated the claim. Best-effort: an unregistered -// provider or a List error leaves the id empty (the backstop re-derives it by -// claim name), so this never fails the caller. -func (r *NodeClaimReconciler) recordInstanceID(ctx context.Context, nc *nebulav1alpha1.NodeClaim) bool { - if nc.Status.InstanceID != "" { - return false - } - prov, ok := r.provider(nc.Spec.Provider) - if !ok { +// recordInstanceID copies the instance id off the served Pod into status.InstanceID when it +// is not already set, returning whether it mutated the claim. +// +// The id comes from the Pod the caller already holds, not from the provider: VK stamps it +// the moment Provision returns (see InstanceIDAnnotation). Asking the provider instead +// meant listing every instance and matching on claim name on EVERY reconcile until the id +// resolved — a remote API call per claim per reconcile, into APIs that rate limit. +// +// Best-effort: an absent annotation means the Pod has not been provisioned yet, or the +// endpoint patch has not landed, so this is a no-op and the Pod watch brings us back when +// it does. Even if the id never arrives, the teardown backstop re-derives it by claim name. +func recordInstanceID(nc *nebulav1alpha1.NodeClaim, pod *corev1.Pod) bool { + if nc.Status.InstanceID != "" || pod == nil { return false } - claim := util.ClaimName(nc.Spec.PodRef.Namespace, nc.Spec.PodRef.Name) - id, err := r.findInstanceID(ctx, prov, claim, "") - if err != nil || id == "" { + id := pod.Annotations[nebulav1alpha1.InstanceIDAnnotation] + if id == "" { return false } nc.Status.InstanceID = id @@ -400,27 +401,32 @@ func (r *NodeClaimReconciler) SetupWithManager(mgr ctrl.Manager) error { For(&nebulav1alpha1.NodeClaim{}). Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(r.claimsForPod)). Named("nodeclaim"). + // One claim per Pod, and the teardown path is the expensive one: per claim a + // self-delete, a provider call to reclaim the instance, and a finalizer update, + // each a round trip. Serialized, that made drain the slowest stage of the + // benchmark by a wide margin (see concurrentReconciles). + WithOptions(controller.Options{MaxConcurrentReconciles: concurrentReconciles}). Complete(r) } -// claimsForPod maps a Pod event to the NodeClaims that serve it (by PodRef). -func (r *NodeClaimReconciler) claimsForPod(ctx context.Context, obj client.Object) []reconcile.Request { +// claimsForPod maps a Pod event to the claim that serves it. The name is DERIVED rather +// than searched for: util.ClaimName is what ensureClaim named the claim, so the same +// namespace/name yields the same token, truncate-and-hash case included. +// +// This used to List every NodeClaim and filter on PodRef, which cost a full cluster-wide +// list plus a deep copy per claim on EVERY Pod event — and the poll loop re-emits every +// tracked pod each tick, so at 500 replicas that was 500 events × 500 copies per tick, +// all of it in this event-handler goroutine, ahead of the workqueue where extra workers +// cannot help. Deriving the name is O(1). +// +// Reconcile still verifies PodRef and the UID pin, so enqueueing a claim that turns out +// to name a different Pod incarnation is harmless. +func (r *NodeClaimReconciler) claimsForPod(_ context.Context, obj client.Object) []reconcile.Request { pod, ok := obj.(*corev1.Pod) if !ok { return nil } - var claims nebulav1alpha1.NodeClaimList - if err := r.List(ctx, &claims); err != nil { - return nil - } - var reqs []reconcile.Request - for i := range claims.Items { - ref := claims.Items[i].Spec.PodRef - if ref.Namespace == pod.Namespace && ref.Name == pod.Name { - reqs = append(reqs, reconcile.Request{ - NamespacedName: types.NamespacedName{Name: claims.Items[i].Name}, - }) - } - } - return reqs + return []reconcile.Request{{ + NamespacedName: types.NamespacedName{Name: util.ClaimName(pod.Namespace, pod.Name)}, + }} } diff --git a/internal/controller/nodeclaim_controller_test.go b/internal/controller/nodeclaim_controller_test.go index 552f3b6..f3ed23c 100644 --- a/internal/controller/nodeclaim_controller_test.go +++ b/internal/controller/nodeclaim_controller_test.go @@ -33,6 +33,7 @@ import ( nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" "github.com/InftyAI/Nebula/pkg/provider" + "github.com/InftyAI/Nebula/pkg/util" ) // fakeProvider is a minimal provider.Provider. On the happy path the NodeClaim @@ -145,6 +146,15 @@ func newPod(name, ns, uid string, phase corev1.PodPhase) *corev1.Pod { } } +// withInstanceID stamps the annotation the virtual kubelet writes once Provision returns, +// which is where the claim controller reads the id from (see recordInstanceID). +func withInstanceID(pod *corev1.Pod, id string) { + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + pod.Annotations[nebulav1alpha1.InstanceIDAnnotation] = id +} + // newClaim builds a steady-state claim: it already carries the terminate // finalizer, since the normal reconcile adds that before doing anything else. // Use newClaimNoFinalizer to exercise the finalizer-addition path itself. @@ -226,12 +236,9 @@ func TestReconcile_PendingPodDoesNotMarkBound(t *testing.T) { // flip the claim to Bound — Bound is the teardown guard, earned only once the // instance is actually running. The instance id is still captured early. pod := newPod("p1", "default", "uid-1", corev1.PodPending) + withInstanceID(pod, "inst-1") claim := newClaim("c1", "p1", "default", "uid-1", "fake") - prov := &fakeProvider{ - name: "fake", - list: []provider.Instance{{ID: "inst-1", ClaimName: "default-p1"}}, - } - r, c := newClaimReconciler(t, []client.Object{pod, claim}, prov) + r, c := newClaimReconciler(t, []client.Object{pod, claim}) reconcileClaim(t, r, "c1") @@ -257,12 +264,9 @@ func TestReconcile_BootingPodEarnsBound(t *testing.T) { // immediately rather than waiting out the grace window. pod := newPod("p1", "default", "uid-1", corev1.PodPending) pod.Status.Reason = podReasonInitializing + withInstanceID(pod, "inst-1") claim := newClaim("c1", "p1", "default", "uid-1", "fake") - prov := &fakeProvider{ - name: "fake", - list: []provider.Instance{{ID: "inst-1", ClaimName: "default-p1"}}, - } - r, c := newClaimReconciler(t, []client.Object{pod, claim}, prov) + r, c := newClaimReconciler(t, []client.Object{pod, claim}) reconcileClaim(t, r, "c1") @@ -324,15 +328,13 @@ func TestReconcile_BoundClaimDoesNotDowngradeOnStatusFlap(t *testing.T) { func TestReconcile_RecordsInstanceIDOnBound(t *testing.T) { // When the served Pod is running, the claim is marked Bound AND its - // status.InstanceID is captured (resolved by claim name via List) so the - // teardown backstop can reclaim the instance even if VK forgets the id. + // status.InstanceID is copied off the Pod annotation VK stamped, so the teardown + // backstop can reclaim the instance even if VK forgets the id. No provider is + // registered here on purpose: the id must come from the Pod, not from a List. pod := newPod("p1", "default", "uid-1", corev1.PodRunning) + withInstanceID(pod, "inst-1") claim := newClaim("c1", "p1", "default", "uid-1", "fake") - prov := &fakeProvider{ - name: "fake", - list: []provider.Instance{{ID: "inst-1", ClaimName: "default-p1"}}, - } - r, c := newClaimReconciler(t, []client.Object{pod, claim}, prov) + r, c := newClaimReconciler(t, []client.Object{pod, claim}) reconcileClaim(t, r, "c1") @@ -384,6 +386,26 @@ func TestReconcile_BoundClaimWithGonePodSelfDeletes(t *testing.T) { } } +func TestReconcile_ProvisioningPodGoneDeletesWithoutGrace(t *testing.T) { + // A claim at Provisioning has already reconciled on a live Pod, so an absent Pod is a + // real teardown, not cache lag: delete at once instead of sitting out the grace window. + claim := newClaim("c1", "pending", "default", "uid-1", "fake") + claim.CreationTimestamp = metav1.NewTime(time.Now()) + claim.Status.Phase = nebulav1alpha1.NodeClaimProvisioning + prov := &fakeProvider{name: "fake"} + r, c := newClaimReconciler(t, []client.Object{claim}, prov) + + res := reconcileClaim(t, r, "c1") + + if res.RequeueAfter > 0 { + t.Fatalf("expected no grace requeue for a claim that observed its Pod, got %+v", res) + } + got := getClaim(t, c, "c1") + if got.DeletionTimestamp.IsZero() { + t.Fatal("expected a Provisioning claim with a gone Pod to be deleted immediately") + } +} + func TestReconcile_NeverObservedPodWaitsForGrace(t *testing.T) { // A claim that never reached Bound and whose Pod is absent is treated as // possible cache lag within the grace window: requeue, do NOT delete. @@ -510,15 +532,16 @@ func TestReconcileDelete_ListErrorRetries(t *testing.T) { } } -func TestClaimsForPod_MapsByPodRef(t *testing.T) { +func TestClaimsForPod_DerivesTheClaimName(t *testing.T) { + // One Pod event enqueues exactly one request, named the way ensureClaim named the + // claim — no cluster-wide List, so the cost does not grow with the number of claims. pod := newPod("p1", "default", "uid-1", corev1.PodRunning) - claim := newClaim("c1", "p1", "default", "uid-1", "fake") - other := newClaim("c2", "other", "default", "uid-2", "fake") - r, _ := newClaimReconciler(t, []client.Object{pod, claim, other}) + other := newClaim(util.ClaimName("default", "other"), "other", "default", "uid-2", "fake") + r, _ := newClaimReconciler(t, []client.Object{pod, other}) reqs := r.claimsForPod(context.Background(), pod) - if len(reqs) != 1 || reqs[0].Name != "c1" { - t.Fatalf("expected only c1 enqueued for pod p1, got %+v", reqs) + if len(reqs) != 1 || reqs[0].Name != util.ClaimName("default", "p1") { + t.Fatalf("expected only the derived claim name for pod p1, got %+v", reqs) } } diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index 192444f..ea4698c 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -27,6 +27,7 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -435,6 +436,9 @@ func (r *PodPlacementReconciler) SetupWithManager(mgr ctrl.Manager) error { For(&corev1.Pod{}). Watches(&nebulav1alpha1.NodePool{}, handler.EnqueueRequestsFromMapFunc(r.podsForPool)). Named("pod-placement"). + // Every opted-in Pod passes through here before the scheduler may touch it, so this + // controller's throughput is the fleet's admission rate (see concurrentReconciles). + WithOptions(controller.Options{MaxConcurrentReconciles: concurrentReconciles}). Complete(r) } diff --git a/internal/controller/sandbox_controller.go b/internal/controller/sandbox_controller.go index d2b4ba0..e5b7b60 100644 --- a/internal/controller/sandbox_controller.go +++ b/internal/controller/sandbox_controller.go @@ -28,6 +28,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" logf "sigs.k8s.io/controller-runtime/pkg/log" @@ -432,5 +433,8 @@ func (r *SandboxReconciler) SetupWithManager(mgr ctrl.Manager) error { For(&nebulav1alpha1.Sandbox{}). Owns(&corev1.Pod{}). Named("sandbox"). + // One Sandbox per box, so a SandboxSet scaled to N puts N objects here (see + // concurrentReconciles). + WithOptions(controller.Options{MaxConcurrentReconciles: concurrentReconciles}). Complete(r) } diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index 7b37a42..e17fc99 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -121,13 +121,16 @@ type trackedPod struct { pod *corev1.Pod claimName string instance string - // connectEndpoint is the endpoint last patched onto the Pod. notify fires every poll - // tick, so this narrows the patch to the ticks where the address actually changed. + // patchedMeta is what VK has already patched onto the Pod in the API server. notify + // fires every poll tick for the pod's whole life, and the pod it hands us is our own + // stamped copy — which always carries these values, so it cannot say whether they ever + // reached etcd. This is that answer, narrowing the patch to the ticks where one + // actually changed (see persistMetadata). // // A cache, not a record: losing it on a restart costs one redundant patch, never a - // lost value, since the annotation lives in etcd. Holds no credential — see + // lost value, since the annotations live in etcd. Holds no credential — see // persistCredential. - connectEndpoint string + patchedMeta podMeta // provisionStart is when THIS process began provisioning. It arms the one // metrics.InstanceReadyDuration observation the poll loop makes on the first @@ -146,6 +149,47 @@ type trackedPod struct { provisionStart time.Time } +// podMeta is the Pod metadata the virtual kubelet owns: the annotations it is the sole +// writer of, both flowing outward to readers (operators, the NodeClaim controller) rather +// than in. Named fields rather than a map keyed by annotation because the set is fixed and +// small, and the compiler should be the one that knows it. +// +// Every field is optional and "" means "VK does not know this yet" — never "clear it". The +// two are learned independently: a provider that mints an address at create knows both at +// once, while one whose address only exists after boot learns the id at create and the +// endpoint from a later poll. +type podMeta struct { + endpoint string + instanceID string +} + +// annotations renders the set fields as a merge-patch body. Unset ones are omitted, so a +// patch can only ever add or update — the annotations are never cleared from here. +func (m podMeta) annotations() map[string]string { + out := make(map[string]string, 2) + if m.endpoint != "" { + out[nebulav1alpha1.EndpointAnnotation] = m.endpoint + } + if m.instanceID != "" { + out[nebulav1alpha1.InstanceIDAnnotation] = m.instanceID + } + return out +} + +// empty reports whether m carries nothing worth writing. +func (m podMeta) empty() bool { return m.endpoint == "" && m.instanceID == "" } + +// minus drops the fields done already holds, leaving only what a patch still has to say. +func (m podMeta) minus(done podMeta) podMeta { + if m.endpoint == done.endpoint { + m.endpoint = "" + } + if m.instanceID == done.instanceID { + m.instanceID = "" + } + return m +} + // NewHandler builds a Handler for one provider backend. The poll cadence comes from // Capabilities.PollInterval, falling back to defaultPollInterval. blocklist (failover // recording) and client (the endpoint patch) may both be nil. @@ -278,6 +322,10 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // copy is what the poll loop re-emits. The emit below publishes it, and every later // tick re-offers it until a write lands. No lock: the Pod is not shared until store. setEndpoint(pod, res.ConnectURL) + // The id travels the same way and for the same reason: stamped before store so the + // tracked copy carries it, published by the emit below, re-offered every tick until a + // write lands. Its reader is the NodeClaim controller (see InstanceIDAnnotation). + setInstanceID(pod, res.InstanceID) h.store(pod, claim, res.InstanceID, provisionStart) // The TOKEN cannot ride the Pod (readable with `get pod`, unencrypted in etcd), so it @@ -472,13 +520,13 @@ func (h *Handler) GetPods(_ context.Context) ([]*corev1.Pod, error) { // // We WRAP VK's callback: its status path writes only the /status subresource and // silently drops metadata changes on the same object, but the endpoint has to live on -// metadata (PodIP cannot hold a DNS name — see applyState). So the wrapper writes the -// access details first, then hands the same Pod to VK. Every status push therefore also -// reconciles how the workload is reached. +// metadata (PodIP cannot hold a DNS name — see applyState) and so does the instance id. So +// the wrapper writes that metadata first, then hands the same Pod to VK. Every status push +// therefore also reconciles how the workload is reached and which instance backs it. func (h *Handler) NotifyPods(ctx context.Context, cb func(*corev1.Pod)) { h.mu.Lock() h.notify = func(pod *corev1.Pod) { - h.persistEndpoint(ctx, pod) + h.persistMetadata(ctx, pod) cb(pod) } h.mu.Unlock() @@ -564,6 +612,10 @@ func (h *Handler) reconcileOnce(ctx context.Context) { // Pending). Re-handing the unchanged status is cheap (dedup, no API write) and arms // VK's own drift correction: the dedup sets lastPodStatusUpdateSkipped, so the next // resync notices the API server disagrees and re-issues the write. + // + // It is NOT free at fleet scale, though: each emit is one rate-limited Enqueue on VK's + // status queue, so N tracked pods offer N items per tick and the queue has to be sized + // above that (see podQueueRate). if notify != nil { for _, p := range emit { notify(p) @@ -590,7 +642,7 @@ func (h *Handler) observeReady(tp *trackedPod, state provider.InstanceState) { } // setEndpoint stamps a reachable address onto the Pod's annotation — the one assignment -// site, wherever the address came from. persistEndpoint then patches it to the API +// site, wherever the address came from. persistMetadata then patches it to the API // server (PodIP cannot hold a DNS name — see applyState). // // Callers, and what each knows: @@ -607,13 +659,31 @@ func (h *Handler) observeReady(tp *trackedPod, state provider.InstanceState) { // // Callers stamping a tracked pod's own Pod must hold h.mu. func setEndpoint(pod *corev1.Pod, endpoint string) { - if endpoint == "" || pod.Annotations[nebulav1alpha1.EndpointAnnotation] == endpoint { + setAnnotation(pod, nebulav1alpha1.EndpointAnnotation, endpoint) +} + +// setInstanceID stamps the provider's instance id on the Pod. One writer only — +// CreatePod, the moment Provision returns it — because that is the only place it is +// learned; the poll loop matches instances by CLAIM NAME and never re-derives the id. +// +// It rides the Pod so the NodeClaim controller can record it from an object it already +// has, instead of asking the provider for a full instance list on every reconcile (see +// InstanceIDAnnotation). Same never-cleared rule as the endpoint. +func setInstanceID(pod *corev1.Pod, id string) { + setAnnotation(pod, nebulav1alpha1.InstanceIDAnnotation, id) +} + +// setAnnotation writes one annotation on the Pod, ignoring an empty value rather than +// clearing what is there, and skipping a value already current. Callers stamping a tracked +// pod's own Pod must hold h.mu. +func setAnnotation(pod *corev1.Pod, key, value string) { + if value == "" || pod.Annotations[key] == value { return } if pod.Annotations == nil { pod.Annotations = map[string]string{} } - pod.Annotations[nebulav1alpha1.EndpointAnnotation] = endpoint + pod.Annotations[key] = value } // statusSignature is a compact rendering of the status fields the poll loop reports, @@ -672,79 +742,88 @@ func (h *Handler) emit(pod *corev1.Pod) { } } -// persistEndpoint writes the address on the emitted Pod to the API server — the write -// half for whichever path stamped it (see setEndpoint). It runs inside the notify wrapper, -// just before VK's status callback, which would drop this metadata change. +// persistMetadata writes the annotations VK owns — the address and the instance id — from +// the emitted Pod to the API server. It is the write half for whichever path stamped them +// (see setEndpoint, setInstanceID), and runs inside the notify wrapper, just before VK's +// status callback, which would drop these metadata changes. +// +// Both in ONE patch: they are usually stamped together by CreatePod, and a second write per +// Pod would be a second round trip on the provisioning path for a value that fits beside +// the first. // -// Because the poll loop re-emits every pod each tick, this is also the retry for any -// failed patch. It carries no credential — a token is written once on the create path. +// Because the poll loop re-emits every pod each tick, this is also the retry for any failed +// patch. It carries no credential — a token is written once on the create path. // // It runs per pod per tick, so anything unconditional is multiplied by the whole fleet; // hence the dedup below. // // Best-effort: a nil client (tests) is a no-op, and a failure is retried next tick. -func (h *Handler) persistEndpoint(ctx context.Context, pod *corev1.Pod) { +func (h *Handler) persistMetadata(ctx context.Context, pod *corev1.Pod) { if h.client == nil { return } - endpoint := pod.Annotations[nebulav1alpha1.EndpointAnnotation] - if endpoint == "" { - return + + want := podMeta{ + endpoint: pod.Annotations[nebulav1alpha1.EndpointAnnotation], + instanceID: pod.Annotations[nebulav1alpha1.InstanceIDAnnotation], } h.mu.Lock() - tp, tracked := h.tracked[key(pod.Namespace, pod.Name)] - // An untracked pod has nothing to dedup against, so patch unconditionally: the - // annotation is the only place the address is published. - patched := false - if tracked { - patched = tp.connectEndpoint == endpoint + // An untracked pod has nothing to compare against, so everything it carries is + // patched: these annotations are the only place those values reach a reader. + if tp, tracked := h.tracked[key(pod.Namespace, pod.Name)]; tracked { + want = want.minus(tp.patchedMeta) } h.mu.Unlock() - if !patched { - h.patchEndpoint(ctx, pod, endpoint) + if !want.empty() { + h.patchMeta(ctx, pod, want) } } -// patchEndpoint merge-patches the endpoint annotation onto the Pod metadata, which needs -// its own write since VK's status callback drops metadata. Scoped to the single -// annotation, so it does not collide with the status write that follows. +// patchMeta merge-patches VK's metadata onto the Pod, which needs its own write since VK's +// status callback drops metadata. Scoped to the annotations want actually carries, so it +// does not collide with the status write that follows. // -// The ONLY write of this annotation, and persistEndpoint its only caller, so every -// address — minted at create or observed at boot — reaches etcd through here. It is never -// called with an empty value, so nothing ever clears it: the annotation is where the -// address lives for the Pod's life. +// The ONLY write of these annotations, and persistMetadata its only caller, so every +// address — minted at create or observed at boot — and every instance id reaches etcd +// through here. Unset fields are omitted rather than sent empty, so nothing here ever +// clears a value: the annotations are where those facts live for the Pod's life. // -// connectEndpoint advances only on success, so a failed patch retries next tick. NotFound -// is ignored — the Pod is gone. -func (h *Handler) patchEndpoint(ctx context.Context, pod *corev1.Pod, endpoint string) { +// patchedMeta advances only on success, so a failed patch retries next tick. NotFound is +// ignored — the Pod is gone. +func (h *Handler) patchMeta(ctx context.Context, pod *corev1.Pod, want podMeta) { patch, err := json.Marshal(map[string]any{ - "metadata": map[string]any{ - "annotations": map[string]string{nebulav1alpha1.EndpointAnnotation: endpoint}, - }, + "metadata": map[string]any{"annotations": want.annotations()}, }) if err != nil { // A fixed-shape map cannot realistically fail to marshal; guard anyway so a future // change surfaces rather than panics. logf.FromContext(ctx).WithName("vnode-handler").Error(err, - "marshal endpoint annotation patch", "pod", key(pod.Namespace, pod.Name)) + "marshal pod annotation patch", "pod", key(pod.Namespace, pod.Name)) return } if _, err := h.client.CoreV1().Pods(pod.Namespace).Patch( ctx, pod.Name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { if !apierrors.IsNotFound(err) { logf.FromContext(ctx).WithName("vnode-handler").Error(err, - "persist endpoint annotation; the poll loop retries next tick", - "pod", key(pod.Namespace, pod.Name), "endpoint", endpoint) + "persist pod annotations; the poll loop retries next tick", + "pod", key(pod.Namespace, pod.Name), "endpoint", want.endpoint, + "instanceID", want.instanceID) } - return // leave connectEndpoint unchanged so the next tick retries + return // leave patchedMeta unchanged so the next tick retries } - // Record success so subsequent ticks skip the patch until the endpoint changes. + // Record success so subsequent ticks skip the patch until a value changes. Only the + // fields this patch carried: an unset one says nothing about what is already there. h.mu.Lock() if tp, ok := h.tracked[key(pod.Namespace, pod.Name)]; ok { - tp.connectEndpoint = endpoint + if want.endpoint != "" { + tp.patchedMeta.endpoint = want.endpoint + } + if want.instanceID != "" { + tp.patchedMeta.instanceID = want.instanceID + } } h.mu.Unlock() } diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index cde152e..5f756be 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -622,7 +622,7 @@ func TestNotify_PersistsEndpointAnnotationOnce(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} h := NewHandler(fp, client, nil) _ = h.CreatePod(context.Background(), pod) - // Register the notify wrapper (this is where persistEndpoint is injected). + // Register the notify wrapper (this is where persistMetadata is injected). h.NotifyPods(context.Background(), func(*corev1.Pod) {}) fp.list = []provider.Instance{{ @@ -651,6 +651,55 @@ func TestNotify_PersistsEndpointAnnotationOnce(t *testing.T) { } } +func TestCreatePod_PersistsInstanceIDAlongsideEndpoint(t *testing.T) { + // The instance id is VK's alone — Provision returns it and nothing re-derives it — so + // it has to reach the Pod for the NodeClaim controller to record status.InstanceID + // without asking the provider for a full instance list on every reconcile. + // + // It rides the SAME patch as the endpoint: both are known the moment Provision returns + // for a provider that mints an address at create, and a second write here would be a + // second round trip per Pod on the provisioning path. + const url = "https://sb-1.modal.host" + pod := testPod("default", "p1") + client := fake.NewSimpleClientset(pod) + + var patches int + client.PrependReactor("patch", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + patches++ + return false, nil, nil // fall through to the tracker so the object updates + }) + + fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} + h := NewHandler(fp, client, nil) + // Wrapper registered BEFORE the create, so the create-path emit is the write. + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get patched pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.InstanceIDAnnotation]; got != "inst-1" { + t.Fatalf("instance id annotation = %q, want inst-1", got) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != url { + t.Fatalf("endpoint annotation = %q, want %q", got, url) + } + if patches != 1 { + t.Fatalf("endpoint and instance id must ride one patch, got %d patches", patches) + } + + // A steady pod is re-emitted every tick with both values unchanged: no further write. + fp.list = []provider.Instance{{ID: "inst-1", ClaimName: "default-p1", State: provider.InstanceRunning}} + h.reconcileOnce(context.Background()) + if patches != 1 { + t.Fatalf("unchanged metadata must not re-patch; got %d patches", patches) + } +} + // connectSecret fetches the connect Secret for a pod, or nil when absent. func connectSecret(t *testing.T, client *fake.Clientset, ns, podName string) *corev1.Secret { t.Helper() @@ -895,7 +944,7 @@ func TestReconcileOnce_EmptyObservedEndpointDoesNotClearAnnotation(t *testing.T) // A create-time URL comes from the provider ONCE and is never re-observed (Modal // reports no endpoint on the read path), so a failed patch cannot be recovered from the // provider. It is recovered from the tracked Pod instead: CreatePod stamps the address -// there, and every poll tick re-emits it, so persistEndpoint keeps patching until one +// there, and every poll tick re-emits it, so persistMetadata keeps patching until one // write lands — then dedups. Without that, one transient 500 at create leaves the // workload permanently unreachable. func TestCreatePod_FailedEndpointPatchIsRetriedByPollLoop(t *testing.T) { diff --git a/pkg/vnode/node.go b/pkg/vnode/node.go index f57f6c0..0b21d07 100644 --- a/pkg/vnode/node.go +++ b/pkg/vnode/node.go @@ -22,6 +22,7 @@ import ( "time" vknode "github.com/virtual-kubelet/virtual-kubelet/node" + "golang.org/x/time/rate" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -30,6 +31,7 @@ import ( "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" logf "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" @@ -45,8 +47,58 @@ 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. Modest, to bound concurrent bursts against a provider's rate limits. +// +// Worthless on its own: the workers pull from queues whose ADMISSION is rate limited, so +// the ceiling is podQueueRate below, not this. See podQueueRateLimiter. const podSyncWorkers = 8 +// podQueueRate and podQueueBurst size the token bucket that admits work into each of the +// pod controller's queues. +// +// This is the single most important number in this file at fleet scale. VK's default, +// which applies whenever the config leaves a limiter nil, is +// workqueue.DefaultControllerRateLimiter(): a bucket of 10 items/s, burst 100, shared by +// EVERY key in the queue. Every Pod event goes in through the rate-limited Enqueue, so +// that default caps the whole node's pod pipeline at 10 events/s no matter how many +// workers run or how much CPU the manager has — and a Pod costs two or three events on +// its way through (the delete, the terminal status it produces, the final removal), so a +// batch teardown measured 3.4-6 Pods/s and did not move when the CPU limit doubled. +// +// 200/s sustained with a 400 burst is deliberately still a ceiling rather than none: the +// queues exist to protect the API server from a fleet-sized burst, and each admitted item +// is a status write. It buys 20x headroom over the default while leaving something in +// place if a bug ever turns into an emit storm. +// +// The floor it has to clear is the poll loop, which re-emits every tracked pod on every +// tick to keep status propagation level-triggered (see reconcileOnce): N pods on the +// default 15s cadence offer N/15 items per second all by themselves — 66/s at a thousand +// pods, which is why the default bucket was permanently saturated at that size. +const ( + podQueueRate = 200 + podQueueBurst = 400 +) + +// podQueueRateLimiter returns the limiter for ONE of the pod controller's queues. +// +// A fresh instance per call, never a shared one: the bucket is per limiter, so handing the +// same value to two queues would make them compete for one budget — the exact coupling +// (status pushes starving deletes) that makes a stall hard to read. +// +// The per-key exponential half is kept exactly as VK's default has it. It only fires after +// a FAILED sync, and it is what stops one pod whose write keeps erroring from spinning; it +// was never the throughput limit, so there is nothing to gain by touching it. +// The TYPED interface, though VK's config field is the untyped workqueue.RateLimiter: +// client-go 0.33 deprecates the latter, and since it is defined as TypedRateLimiter[any] the +// value returned here still assigns to that field. +func podQueueRateLimiter() workqueue.TypedRateLimiter[any] { + return workqueue.NewTypedMaxOfRateLimiter[any]( + workqueue.NewTypedItemExponentialFailureRateLimiter[any](5*time.Millisecond, 1000*time.Second), + &workqueue.TypedBucketRateLimiter[any]{ + Limiter: rate.NewLimiter(rate.Limit(podQueueRate), podQueueBurst), + }, + ) +} + // NodeName returns the virtual node name for a provider: "nebula-". // One static node per provider (see docs/architecture.md §3); the scheduler // routes an ungated Pod to it via the ProviderLabel nodeSelector. @@ -146,6 +198,13 @@ func (r *Runner) Start(ctx context.Context) error { SecretInformer: secretInformer, ConfigMapInformer: configMapInformer, ServiceInformer: serviceInformer, + // All three set EXPLICITLY, because a nil one silently gets a 10 items/s bucket + // (see podQueueRateLimiter). Each gets its own limiter so the three cannot starve + // each other: work arrives here from Kubernetes, status arrives from the provider, + // and the deletes are what a teardown waits on. + SyncPodsFromKubernetesRateLimiter: podQueueRateLimiter(), + SyncPodStatusFromProviderRateLimiter: podQueueRateLimiter(), + DeletePodsFromKubernetesRateLimiter: podQueueRateLimiter(), }) if err != nil { return fmt.Errorf("build pod controller for %q: %w", r.nodeName, err) diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index 72552b1..ef5f236 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -518,6 +518,61 @@ type tokenRequest struct { } `json:"status"` } +// waitForControllerReady blocks until the manager can serve the whole path a new Pod +// takes: the Deployment reports Available, the mutating webhook carries the CA bundle the +// manager mints for itself, and the fake provider's virtual node reports Ready. +// +// It exists because `make test-perf` filters the suite to the perf label, so the specs that +// assert the manager came up ("should run successfully", "should have CA injection") never +// run — and a batch applied against a half-started manager is not a failed run, it is a +// WRONG one. Both gaps inflate a stage that then reads as Nebula being slow: +// +// - the webhook is failurePolicy=Fail on Pod CREATE, so until pkg/cert has patched its +// caBundle from inside the manager, every replica create is REJECTED and the ReplicaSet +// retries with backoff. That lands in "Pods creation". +// - the Node object exists a moment before the node controller marks it Ready, and while +// it is NotReady the node lifecycle controller taints it NoSchedule, so an ungated Pod +// cannot bind. That lands in "Pods bound". +// +// Waiting on the CONDITION rather than on the pod phase for the first check: Running is +// true of a manager whose probes have not passed and which has not won leader election, and +// leader election is what gates every reconcile the benchmark measures. +func waitForControllerReady() { + By("waiting for the controller-manager Deployment to report Available") + // By label, not by name, so this does not have to know the kustomize name prefix. An + // Eventually over `get` rather than `kubectl wait`, because `wait` on a selector that + // matches nothing yet fails outright instead of retrying. + Eventually(func(g Gomega) { + out, err := utils.Run(exec.Command("kubectl", "get", "deployment", + "-l", "control-plane=controller-manager", "-n", namespace, + "-o", "go-template={{range .items}}{{range .status.conditions}}"+ + "{{if eq .type \"Available\"}}{{.status}}{{end}}{{end}}{{end}}")) + g.Expect(err).NotTo(HaveOccurred(), "Failed to read the controller-manager Deployment") + g.Expect(out).To(Equal("True"), "the controller-manager is not Available yet") + }).Should(Succeed()) + + By("waiting for the mutating webhook's CA bundle to be injected") + Eventually(func(g Gomega) { + out, err := utils.Run(exec.Command("kubectl", "get", + "mutatingwebhookconfigurations.admissionregistration.k8s.io", + "nebula-mutating-webhook-configuration", + "-o", "go-template={{range .webhooks}}{{.clientConfig.caBundle}}{{end}}")) + g.Expect(err).NotTo(HaveOccurred()) + // Same threshold as the CA-injection spec: enough to tell a real bundle from an + // empty field, without pinning the cert's size. + g.Expect(len(out)).To(BeNumerically(">", 10), "the webhook caBundle is not injected yet") + }).Should(Succeed()) + + By("waiting for the fake provider's virtual node to report Ready") + Eventually(func(g Gomega) { + out, err := utils.Run(exec.Command("kubectl", "get", "node", fakeVirtualNode, + "-o", "go-template={{range .status.conditions}}"+ + "{{if eq .type \"Ready\"}}{{.status}}{{end}}{{end}}")) + g.Expect(err).NotTo(HaveOccurred(), "fake virtual node not registered") + g.Expect(out).To(Equal("True"), "fake virtual node registered but not Ready") + }).Should(Succeed()) +} + // waitForNodeClaimsGone polls until no NodeClaim is left, reporting whether they // drained within timeout. A leftover claim means its terminate finalizer never ran, so // the caller must not delete the CRD (it would wedge) and an instance may be leaking. diff --git a/test/e2e/perf_report_test.go b/test/e2e/perf_report_test.go index ffb8d6f..f6d76f4 100644 --- a/test/e2e/perf_report_test.go +++ b/test/e2e/perf_report_test.go @@ -31,10 +31,9 @@ import ( "github.com/InftyAI/Nebula/test/utils" ) -// The HTML report is deliberately ONE self-contained file: no CDN, no JavaScript, and -// the chart is inline SVG computed here. A perf report gets opened from a laptop with no -// network, mailed around, and attached to CI artifacts, and any of those breaks the -// moment it needs to fetch a charting library. +// The HTML report is deliberately ONE self-contained file: no CDN, no JavaScript, the chart +// inline SVG computed here. It gets opened offline, mailed around, and attached to CI +// artifacts, and a fetched charting library breaks all three. const ( // perfReportDir sits under the repo root, which .gitignore already covers // ("artifacts"), so reports cannot be committed by accident. @@ -42,15 +41,15 @@ const ( perfReportFile = "perf-report.html" ) -// Stage colours are shared by the table swatches and the chart curves, so a row and its -// curve are the same colour and the eye can move between them. Two colours do double duty, -// once per section, so they never share a chart: colourCreated for "Pods creation" and "Pods -// gone" (the Pod object moving at Kubernetes' own pace) and colourSync for the two derived -// per-workload rows, which are the only rows that isolate Nebula's own cost. +// Shared by the table swatches and the chart curves, so a row and its curve match. Two +// reuses, both safe: colourCreated appears once per section and the sections never share a +// chart, and the derived per-claim row takes colourClaims to sit beside the claims curve it +// is computed from — a derived row draws no curve, so it cannot make two curves ambiguous. const ( colourCreated = "#94a3b8" colourClaims = "#f59e0b" colourBound = "#3b82f6" + colourReady = "#8b5cf6" colourSync = "#10b981" colourGone = "#ef4444" ) @@ -125,18 +124,14 @@ type reportSeries struct { Label string Colour string Points string // SVG polyline "x,y x,y ..." - // Dash makes an exactly-covered curve visible without moving or fattening anything. - // Two stages CAN be identical, point for point — a Pod already bound the first time the - // poller sees it stamps "created" and "bound" from one timestamp — and the curve drawn - // first would otherwise vanish completely, which reads as missing data rather than as - // agreement. So the curve on TOP is dashed and the one underneath shows through the - // gaps: same width, same path, just interrupted. + // Dash makes an exactly-covered curve visible. Two stages CAN be identical point for point + // — a Pod already bound when first polled stamps created and bound from one timestamp — and + // the curve drawn first would then vanish, which reads as missing data rather than as + // agreement. So the curve on TOP is dashed and the one beneath shows through the gaps. // - // Drawing the covered curve as a wide translucent band was the previous attempt. These - // curves are staircases, and a 9px stroke on a near-vertical run protrudes to both - // sides of the 2.25px line on top of it, which reads as a second line offset sideways — - // the width itself became a misleading signal. Nudging a curve off its real position was - // never an option: that would be a lie about the numbers. + // Never widen or nudge instead: on these staircases a 9px band protrudes either side of the + // 2.25px line and reads as a second, offset curve, and moving a curve would misstate the + // numbers. Dash string // stroke-dasharray; empty for a solid curve // Same names the curve this one duplicates, for the legend. Same string @@ -175,47 +170,46 @@ type reportPage struct { type reportCard struct { Key string Value string - // Note becomes the card's tooltip. A six-card strip has no room for prose, but a - // two-word key does not say what the number covers — "total" alone does not reveal - // that cluster setup is outside it — so the definition has to live somewhere. + // Note becomes the card's tooltip: the strip has no room for prose, and a two-word key + // cannot say what the number covers ("total" does not reveal that setup is outside it). Note string } func renderHTMLReport( n int, s syncSamples, total, drainTotal time.Duration, d drainSamples, ) (string, error) { - // Measured from the apply. Ordered as the path runs, not by duration. + // Pods first, then the claim ledger; within each group the cumulative rows (from the apply, + // stamped only once the manager is ready — see waitForControllerReady) before the derived + // per-object one, which divides the arrival rate out and so survives a change of batch size. syncStages := []stage{{ Label: "Pods creation", Colour: colourCreated, Samples: s.created, Total: n, Curve: true, Note: "Kubernetes' own cost: how fast the Pods are created.", }, { Label: "Pods bound to " + fakeVirtualNode, Colour: colourBound, Samples: s.bound, Total: n, Curve: true, - Note: "Ungated and scheduled onto the virtual node — the workload is live.", + Note: "Ungated and scheduled onto the virtual node.", + }, { + Label: "Pods Ready", Colour: colourReady, Samples: s.ready, Total: n, Curve: true, + Note: "End to end: the workloads are usable. The batch is deleted only after this.", + }, { + Label: "Placement (Pod created → bound)", Colour: colourSync, Samples: s.placement, + Note: "Per Pod: the gate coming off plus the bind, arrival rate factored out.", }, { Label: "NodeClaims Bound", Colour: colourClaims, Samples: s.claims, Total: n, Curve: true, Note: "The claim ledger catching up: placement decided and recorded.", }, { - Label: "Per-Pod sync (created → bound)", Colour: colourSync, Samples: s.sync, - Note: "Nebula's own contribution, with the creation rate factored out.", + Label: "Provisioning (NodeClaim created → Bound)", Colour: colourClaims, Samples: s.provision, + Note: "Per claim: created until an instance exists. Contains the bind above, because a " + + "claim goes Bound off its Pod's status.", }} - // Measured from the delete, so these get their own clock — and their own chart. Pods - // first because the delete lands on the workload first, not because they finish first: - // which curve trails is the result, and it has gone both ways. + // Measured from the delete, so these get their own clock and their own chart. Pods first + // because the delete lands there first, not because they finish first — which curve trails + // is the result, and it has gone both ways. drainStages := []stage{{ Label: "Pods gone", Colour: colourCreated, Samples: d.podsGone, Total: d.podsKnown, Curve: true, Note: "Graceful termination through the virtual kubelet.", }, { Label: "NodeClaims gone", Colour: colourGone, Samples: d.gone, Total: d.known, Curve: true, - Note: "Self-deleted once the served Pod is gone, then the terminate finalizer releases " + - "the instance. Both rows count against the batch the sync watch observed; anything " + - "already gone at the first drain poll is stamped there, an upper bound.", - }, { - Label: "Per-claim release (pod → claim gone)", Colour: colourSync, Samples: d.release, - Total: d.podsKnown, - Note: "Nebula's own contribution to teardown, with the Pod deletion rate factored out. " + - "The two curves above cannot show this: their percentiles are over different objects. " + - "Both stamps come from one snapshot, so most pairs land inside a single poll and read " + - "as 0. A count below the batch is pairs with no end yet, or a claim seen gone first.", + Note: "Self-deleted once its Pod is gone; the terminate finalizer then releases the instance.", }} page := reportPage{ @@ -223,18 +217,25 @@ func renderHTMLReport( Generated: time.Now().Format("Mon, 02 Jan 2006 15:04:05 MST"), Cards: []reportCard{ {Key: "total", Value: fmtDuration(total), - Note: "The whole benchmark: apply until the last NodeClaim was gone, so sync plus " + - "teardown plus the short gap where the sync numbers are reported and asserted. " + - "Cluster setup and the manager's deploy are outside it."}, - {Key: "all synced", Value: lastOf(s.bound, n), - Note: "From the apply until the last Pod was bound to the virtual node. The apply " + - "itself is not reported: its round trip is the client's cost, not Nebula's."}, - {Key: "teardown", Value: fmtDuration(drainTotal), - Note: "From the delete until every NodeClaim in the batch was gone."}, + Note: "Apply until the last NodeClaim was gone: sync, teardown, and the gap where " + + "the sync numbers are asserted. Cluster setup and the deploy are outside it. On " + + "a run that gave up, apply until it gave up — read it against the verdict."}, {Key: "replicas", Value: fmt.Sprintf("%d", n), Note: "Batch size. Override with NEBULA_E2E_PERF_WORKLOADS."}, - {Key: "sync throughput", Value: rate(len(s.bound), n, lastSample(s.bound), "workloads/s"), - Note: "Replicas bound per second across that window."}, + {Key: "all pods ready", Value: lastOf(s.ready, n), + Note: "Apply until the last Pod reported Ready: how long until N workloads are " + + "usable. Never ahead of all pods bound, and the batch is deleted only after this."}, + // Named for the OBJECT: a Pod is bound to a node by the scheduler, while Bound is a + // NodeClaim PHASE already satisfied while the instance initializes (see desiredPhase). + {Key: "all pods bound", Value: lastOf(s.bound, n), + Note: "Apply until the last Pod was bound to the virtual node — the scheduling " + + "decision, not usability (all pods ready) and not the claim ledger " + + "(NodeClaims Bound)."}, + {Key: "bind rate", Value: rate(len(s.bound), n, lastSample(s.bound), "workloads/s"), + Note: "Pods bound to the virtual node per second across that window."}, + {Key: "teardown", Value: fmtDuration(drainTotal), + Note: "Delete until every NodeClaim in the batch was gone. Empty when the sync never " + + "finished, since nothing was deleted — unlike total, stamped either way."}, {Key: "drain rate", Value: rate(len(d.gone), len(d.gone), drainTotal, "claims/s"), Note: "NodeClaims removed per second, finalizers included."}, }, @@ -254,9 +255,11 @@ func renderHTMLReport( page.DrainChart = buildChart(drainStages, max(d.known, d.podsKnown), "elapsed since delete") switch { - case len(s.bound) < n || len(s.claims) < n: - page.Verdict = fmt.Sprintf("INCOMPLETE — %d/%d pods bound, %d/%d claims Bound%s", - len(s.bound), n, len(s.claims), n, stalledSuffix(s.stalled)) + case len(s.bound) < n || len(s.claims) < n || len(s.ready) < n: + // Ready named alongside the other two, not folded in: "bound but never Ready" is a + // provisioning or status problem, "never bound" is a placement one. + page.Verdict = fmt.Sprintf("INCOMPLETE — %d/%d pods bound, %d/%d claims Bound, %d/%d pods Ready%s", + len(s.bound), n, len(s.claims), n, len(s.ready), n, stalledSuffix(s.stalled)) page.VerdictBad = true case d.remaining > 0: page.Verdict = fmt.Sprintf("Synced, but %d claim(s) never drained%s", @@ -305,9 +308,8 @@ func rate(count, want int, over time.Duration, unit string) string { } func buildRows(stages []stage) []reportRow { - // Bars are scaled within a table, never across the two: the sync stages and the - // drain are measured from different starts, so a shared scale would invite a - // comparison that means nothing. + // Bars scale within a table, never across the two: sync and teardown are measured from + // different starts, so a shared scale would invite a meaningless comparison. scale := time.Duration(0) for _, st := range stages { if v := lastSample(st.Samples); v > scale { @@ -322,8 +324,8 @@ func buildRows(stages []stage) []reportRow { case len(st.Samples) == 0: row.Count, row.P50, row.P95, row.Max, row.Muted = "—", "—", "—", "—", true case lastSample(st.Samples) == 0: - // Same guard as the terminal report: all-zero means faster than one poll, and - // printing 0s would claim precision this measurement does not have. + // As in the terminal report: all-zero means faster than one poll, and printing 0s + // would claim precision this measurement does not have. row.P50, row.P95, row.Max = "< 1 poll", "< 1 poll", "< 1 poll" default: row.P50 = fmtDuration(percentile(st.Samples, 50)) @@ -357,10 +359,10 @@ func fmtDuration(d time.Duration) string { } } -// buildChart turns the samples into cumulative-completion curves, or returns nil if -// there is nothing to draw. Sorted ascending, sample i IS the moment the (i+1)-th -// workload cleared that stage, so no bucketing is needed: steepness is the rate, a flat -// stretch is a stall, and a curve hugging another means that stage is keeping up. +// buildChart turns the samples into cumulative-completion curves, or nil if there is nothing +// to draw. Sorted ascending, sample i IS the moment the (i+1)-th workload cleared the stage, +// so no bucketing: steepness is the rate, a flat stretch is a stall, and a curve hugging +// another means that stage is keeping up. func buildChart(stages []stage, want int, xLabel string) *chart { const w, h, padL, padR, padT, padB = 880, 300, 54, 24, 16, 44 c := &chart{ @@ -393,8 +395,8 @@ func buildChart(stages []stage, want int, xLabel string) *chart { if !st.Curve || len(st.Samples) == 0 { continue } - // Start at (first sample, 0) so a curve that begins late reads as beginning late - // rather than as rising out of the origin. + // Start at (first sample, 0) so a late curve reads as starting late rather than as + // rising out of the origin. pts := make([]string, 0, len(st.Samples)+1) pts = append(pts, fmt.Sprintf("%.1f,%.1f", x(st.Samples[0].Seconds()), y(0))) for i, at := range st.Samples { @@ -408,9 +410,8 @@ func buildChart(stages []stage, want int, xLabel string) *chart { return nil } - // Mark exact overlaps: dash the curve on top so the one beneath shows through it, and - // name the pairing in the legend, so "one line is missing" reads as "these two are the - // same line". Both curves keep the same width and position — see reportSeries.Dash. + // Mark exact overlaps: dash the curve on top so the one beneath shows through, so "a line + // is missing" reads as "these two are identical" — see reportSeries.Dash. for i := range c.Series { for j := 0; j < i; j++ { if c.Series[i].Points == c.Series[j].Points { @@ -433,27 +434,29 @@ func buildChart(stages []stage, want int, xLabel string) *chart { // diffed against another run without re-deriving anything. func plainSummary(n int, s syncSamples, total, drainTotal time.Duration, d drainSamples) string { var b strings.Builder - fmt.Fprintf(&b, "replicas %d\n", n) - fmt.Fprintf(&b, "Pods created %s\n", stageLine(s.created, n)) - fmt.Fprintf(&b, "NodeClaims Bound %s\n", stageLine(s.claims, n)) - fmt.Fprintf(&b, "Pods bound to %-18s %s\n", fakeVirtualNode, stageLine(s.bound, n)) - fmt.Fprintf(&b, "per-Pod sync (created → bound) %s\n", spreadLine(s.sync)) - fmt.Fprintf(&b, "teardown total %s\n", fmtDuration(drainTotal)) - fmt.Fprintf(&b, "Pods gone %s\n", stageLine(d.podsGone, d.podsKnown)) - fmt.Fprintf(&b, "NodeClaims gone %s\n", stageLine(d.gone, d.known)) - fmt.Fprintf(&b, "per-claim release (pod → claim) %s\n", releaseLine(d)) - fmt.Fprintf(&b, "total (apply → drained) %s\n", fmtDuration(total)) - fmt.Fprintf(&b, "poll interval %s\n", perfPollInterval) - fmt.Fprintf(&b, "stall timeout %s\n", perfStallTimeout) - fmt.Fprintf(&b, "sync budget %s\n", perfBudget(n)) - fmt.Fprintf(&b, "teardown budget %s\n", perfTeardownBudget(n)) + // Literal spaces, never %-35s: Printf pads by BYTES and the arrow is three of them, so a + // width verb short-pads exactly the rows carrying one. Sized to the longest label. + fmt.Fprintf(&b, "replicas %d\n", n) + fmt.Fprintf(&b, "Pods created %s\n", stageLine(s.created, n)) + fmt.Fprintf(&b, "Pods bound to %-18s %s\n", fakeVirtualNode, stageLine(s.bound, n)) + fmt.Fprintf(&b, "Pods Ready %s\n", stageLine(s.ready, n)) + fmt.Fprintf(&b, "placement (Pod created → bound) %s\n", spreadLine(s.placement)) + fmt.Fprintf(&b, "NodeClaims Bound %s\n", stageLine(s.claims, n)) + fmt.Fprintf(&b, "provisioning (NC created → Bound) %s\n", spreadLine(s.provision)) + fmt.Fprintf(&b, "teardown total %s\n", fmtDuration(drainTotal)) + fmt.Fprintf(&b, "Pods gone %s\n", stageLine(d.podsGone, d.podsKnown)) + fmt.Fprintf(&b, "NodeClaims gone %s\n", stageLine(d.gone, d.known)) + fmt.Fprintf(&b, "total (apply → drained) %s\n", fmtDuration(total)) + fmt.Fprintf(&b, "poll interval %s\n", perfPollInterval) + fmt.Fprintf(&b, "stall timeout %s\n", perfStallTimeout) + fmt.Fprintf(&b, "sync budget %s\n", perfBudget(n)) + fmt.Fprintf(&b, "teardown budget %s\n", perfTeardownBudget(n)) return b.String() } -// html/template escapes every interpolation, so nothing computed above can break the -// page even if a label or provider name ever carries markup. Two template blocks are -// shared by the sync and teardown sections, which is what keeps the two honest about -// being the same measurement on different clocks. +// html/template escapes every interpolation, so no label or provider name can break the page. +// The sync and teardown sections share both template blocks, which is what keeps them honest +// about being the same measurement on different clocks. var reportTemplate = template.Must(template.New("perf").Parse(` @@ -506,13 +509,22 @@ var reportTemplate = template.Must(template.New("perf").Parse(` .verdict .dot { width: 7px; height: 7px; border-radius: 50%; background: currentColor; flex: none; transform: translateY(-3px); } /* Hairline grid: 1px gaps filled by the container's background, so the cards read as one - block instead of six floating boxes. */ - .cards { display: grid; grid-template-columns: repeat(3, 1fr); gap: 1px; + block. Flex that GROWS, not fixed grid columns — a short last row's empty cells would + paint that hairline as a grey slab, while flex has the row share the width instead. + + Two bases put THREE on the top row: the first three take 30% (3x30% + 22% > 100%, so the + fourth wraps), the rest 22% (4x22% < 100% < 5x22%). Seven cards land 3 + 4. */ + .cards { display: flex; flex-wrap: wrap; gap: 1px; background: var(--line); border: 1px solid var(--line); border-radius: var(--radius); overflow: hidden; box-shadow: var(--shadow); margin-top: 18px; } - @media (max-width: 640px) { .cards { grid-template-columns: repeat(2, 1fr); } } - .card { background: var(--panel); padding: 12px 15px 14px; } + .card { flex: 1 1 22%; background: var(--panel); padding: 12px 15px 14px; } + .cards .card:nth-child(-n+3) { flex-basis: 30%; } + /* Two per row on a phone. Both selectors: the nth-child rule above outranks a bare .card, + and would otherwise hold the headline row at three. */ + @media (max-width: 640px) { + .card, .cards .card:nth-child(-n+3) { flex-basis: 45%; } + } .card .k { color: var(--faint); font-size: 9.5px; font-weight: 580; text-transform: uppercase; letter-spacing: .08em; } .card .v { font-size: 17px; font-weight: 600; font-variant-numeric: tabular-nums; @@ -578,7 +590,7 @@ var reportTemplate = template.Must(template.New("perf").Parse(`
-

Workload sync benchmark

+

Workload Benchmark

{{.Generated}}

Fake provider on Kind, so this is control-plane cost only: Pods bind to a @@ -614,8 +626,7 @@ var reportTemplate = template.Must(template.New("perf").Parse(`

Generated by make test-perf (test/e2e/perf_test.go). Set NEBULA_E2E_PERF_WORKLOADS to change the batch size and - NEBULA_E2E_PERF_REPORT to move this file — two runs sharing the - default path overwrite each other. + NEBULA_E2E_PERF_REPORT.
@@ -652,7 +663,7 @@ var reportTemplate = template.Must(template.New("perf").Parse(` {{define "chart"}}
+ aria-label="cumulative Pods past each stage, {{.XLabel}}"> {{range .YTicks}} {{.Label}} @@ -662,14 +673,12 @@ var reportTemplate = template.Must(template.New("perf").Parse(` {{end}} {{.XLabel}} - - workloads - + + Pods + {{range .Series}} collapse relies on the + // same rule). Both facts then also come off the same object at the same instant. out, err = utils.RunQuiet(exec.Command("kubectl", "get", "pods", "-n", perfWorkloadNS, - "-o", "go-template={{range .items}}{{.metadata.name}} {{.spec.nodeName}}{{\"\\n\"}}{{end}}")) + "-o", "go-template={{range .items}}{{.metadata.name}} {{.spec.nodeName}}|"+ + "{{range .status.conditions}}{{if eq .type \"Ready\"}}{{.status}}{{end}}{{end}}"+ + "{{\"\\n\"}}{{end}}")) if err == nil { at := time.Since(start) - for name, node := range batchRows(out, perfDeployName) { + for name, val := range batchRows(out, perfDeployName) { + node, ready := podRow(val) firstSeen(createdSeen, name, at) if node == fakeVirtualNode { firstSeen(boundSeen, name, at) } + // The literal the Ready condition carries when the vnode has observed the + // instance running (see pkg/vnode setReady). A Pod with no conditions yet + // renders as empty, which matches nothing. + if ready == "True" { + firstSeen(readySeen, name, at) + } } } - if len(createdSeen) >= n && len(claimSeen) >= n && len(boundSeen) >= n { + if len(createdSeen) >= n && len(claimSeen) >= n && len(boundSeen) >= n && len(readySeen) >= n { break } - if got := len(createdSeen) + len(claimSeen) + len(boundSeen); got > progress { + if got := len(createdSeen) + len(claimSeen) + len(boundSeen) + len(readySeen); got > progress { progress, lastProgress = got, time.Now() } if time.Since(lastProgress) > perfStallTimeout { @@ -298,20 +360,27 @@ func watchBatchSync(n int, start time.Time, budget time.Duration) syncSamples { // the Pod's .spec.nodeName are different objects reaching different states, and // one line carrying "bound" for both reads as a single stage counted twice. _, _ = fmt.Fprintf(GinkgoWriter, - " t=%s pods created %d/%d claims Bound %d/%d pods on %s %d/%d\n", + " t=%s pods created %d/%d claims Bound %d/%d pods on %s %d/%d pods Ready %d/%d\n", time.Since(start).Round(time.Second), len(createdSeen), n, len(claimSeen), n, - fakeVirtualNode, len(boundSeen), n) + fakeVirtualNode, len(boundSeen), n, len(readySeen), n) nextLog = time.Now().Add(10 * time.Second) } time.Sleep(perfPollInterval) } - // Only Pods observed at both ends contribute a sync sample; a Pod still unbound - // has no end yet, and counting it as zero would flatter the result. - sync := make([]time.Duration, 0, len(boundSeen)) + // Only objects observed at BOTH ends contribute: one still short of its end has no window + // yet, and counting it as zero would flatter the result. So on a run that gave up these + // cover only the subset that finished, which is why the spread rows show no count. + placement := make([]time.Duration, 0, len(boundSeen)) for name, at := range boundSeen { if c, ok := createdSeen[name]; ok { - sync = append(sync, at-c) + placement = append(placement, at-c) + } + } + provision := make([]time.Duration, 0, len(claimSeen)) + for name, at := range claimSeen { + if c, ok := claimCreatedSeen[name]; ok { + provision = append(provision, at-c) } } @@ -321,12 +390,14 @@ func watchBatchSync(n int, start time.Time, budget time.Duration) syncSamples { } return syncSamples{ - created: ascending(createdSeen), - claims: ascending(claimSeen), - bound: ascending(boundSeen), - sync: sortDurations(sync), - podNames: names, - stalled: stalled, + created: ascending(createdSeen), + claims: ascending(claimSeen), + bound: ascending(boundSeen), + ready: ascending(readySeen), + placement: sortDurations(placement), + provision: sortDurations(provision), + podNames: names, + stalled: stalled, } } @@ -345,28 +416,13 @@ type drainSamples struct { // podsGone is the same measurement for this batch's Pods, on the same clock. It is what // says how much of the teardown is Kubernetes' own: the Pod waits on graceful termination // through the virtual kubelet, and the claim cannot go until that finishes, so this curve - // is the floor under the one above. Do not read the ORDER off the two curves — see - // release for that. + // is the floor under the one above. Do not read the ORDER off the two curves: their + // percentiles are over different objects, so a claim at p50 and a Pod at p50 are not the + // same workload. // // Counted against podsKnown, seeded and stamped exactly as gone is. podsGone []time.Duration podsKnown int - // release pairs each Pod with its OWN claim: claim gone − pod gone, ascending, both stamped - // from the same snapshot. This is the teardown twin of syncSamples.sync, and the only number - // here that isolates Nebula: the two curves above are dominated by however fast the virtual - // kubelet processes 500 pod deletions, while this one says what the claim path costs on top - // of that. Expect most of it to read as 0 — the claim follows well inside one poll. - // - // Comparing the two distributions cannot answer that — their percentiles are over - // different objects, so a claim at p50 and a Pod at p50 are not the same workload, and a - // few missing Pod samples can make the claims look like they went first. - release []time.Duration - // releaseOutOfOrder counts pairs left OUT of release because the claim was seen absent while - // its Pod was still listed. With one snapshot per poll that is a genuine inversion rather - // than a sampling artifact, so it should read 0; anything else means a claim outran the Pod - // it serves. Dropped rather than kept as a negative sample, and counted rather than dropped - // quietly, because an inversion needs somewhere to show up. - releaseOutOfOrder int // remaining is how many of this batch's claims were still present when the poll // gave up; 0 means drained. remaining int @@ -379,13 +435,11 @@ type drainSamples struct { // above it fails fast on a stall — a count falling steadily is progress however slow, // while a count that stops falling is wedged. // -// ONE list call per poll, covering Pods and claims together, unlike the sync watch. Two -// calls made the per-workload pairing unmeasurable: a claim follows its Pod in tens of -// milliseconds (see NodeClaimReconciler.Reconcile — it self-deletes once the served Pod -// reads absent), which is no bigger than the gap between two kubectl invocations, so every -// retained delta was measuring the harness rather than Nebula and roughly a third of the -// batch came out inverted and dropped. One snapshot stamps a Pod and its claim from the same -// instant. See pairRelease. +// ONE list call per poll, covering Pods and claims together, unlike the sync watch. A claim +// follows the Pod it serves in tens of milliseconds (see NodeClaimReconciler.Reconcile — it +// self-deletes once the served Pod reads absent), which is smaller than the gap between two +// kubectl invocations, so two calls put the two curves on measurably different clocks and made +// their order unreadable. One snapshot stamps a Pod and its claim from the same instant. // // podNames is the batch the sync watch observed (syncSamples.podNames). Both ledgers are // seeded from it so the counts are against the batch that provably existed, not against @@ -400,8 +454,8 @@ func drainPerfClaims(budget time.Duration, podNames []string) drainSamples { start := time.Now() deadline := start.Add(budget) - known := map[string]struct{}{} - goneSeen := map[string]time.Duration{} + claimsKnown := map[string]struct{}{} + claimsGoneSeen := map[string]time.Duration{} podsKnown := map[string]struct{}{} podsGoneSeen := map[string]time.Duration{} // Seeding only asserts these objects EXISTED. The Pods were observed directly by the sync @@ -411,7 +465,7 @@ func drainPerfClaims(budget time.Duration, podNames []string) drainSamples { // first poll that fails to see it, the same rule every other name follows. for _, pod := range podNames { podsKnown[pod] = struct{}{} - known[nebulautil.ClaimName(perfWorkloadNS, pod)] = struct{}{} + claimsKnown[nebulautil.ClaimName(perfWorkloadNS, pod)] = struct{}{} } remaining := -1 // no observation yet, so the first one always counts as progress lastProgress := start @@ -433,21 +487,18 @@ func drainPerfClaims(budget time.Duration, podNames []string) drainSamples { at := time.Since(start) observeGone(podsKnown, podsGoneSeen, batchRows(out, perfDeployName), at) claims := batchRows(out, claimPrefix) - observeGone(known, goneSeen, claims, at) + observeGone(claimsKnown, claimsGoneSeen, claims, at) return len(claims), true } result := func(remaining int, stalled bool) drainSamples { - release, outOfOrder := pairRelease(podsGoneSeen, goneSeen) return drainSamples{ - gone: ascending(goneSeen), - known: len(known), - podsGone: ascending(podsGoneSeen), - podsKnown: len(podsKnown), - release: release, - releaseOutOfOrder: outOfOrder, - remaining: remaining, - stalled: stalled, + gone: ascending(claimsGoneSeen), + known: len(claimsKnown), + podsGone: ascending(podsGoneSeen), + podsKnown: len(podsKnown), + remaining: remaining, + stalled: stalled, } } @@ -473,44 +524,18 @@ func drainPerfClaims(budget time.Duration, podNames []string) drainSamples { } } -// pairRelease derives one sample per workload: how long after its Pod vanished the claim -// serving it followed. Only workloads with BOTH stamps contribute — a claim still holding -// its finalizer has no end yet, and counting it as zero would flatter the result, the same -// rule syncSamples.sync follows at the other end. +// batchRows parses " " lines, keeping those whose name carries prefix. // -// Keyed through ClaimName rather than a hand-rolled join so this matches whatever the -// controller derived, including the truncate-and-hash case for long names. -// -// On sign: both stamps come from the same snapshot (see drainPerfClaims), so a pair that -// vanishes inside one poll reads as 0 rather than as the gap between two list calls, and -// claimAt < podAt can only mean a genuine inversion — the claim was seen absent while its -// Pod was still listed. Those are returned as a count instead of as negative samples, so the -// distribution stays interpretable while nothing is silently discarded. Nothing is clamped: a -// clamp would make a real inversion look like 0. -func pairRelease(podsGoneSeen, goneSeen map[string]time.Duration) ([]time.Duration, int) { - out := make([]time.Duration, 0, len(podsGoneSeen)) - outOfOrder := 0 - for pod, podAt := range podsGoneSeen { - claimAt, ok := goneSeen[nebulautil.ClaimName(perfWorkloadNS, pod)] - switch { - case !ok: - // The claim has no end yet: still holding its finalizer, or the poll gave up first. - // Counting it as zero would flatter the result, the same rule syncSamples.sync follows. - case claimAt < podAt: - outOfOrder++ - default: - out = append(out, claimAt-podAt) - } - } - return sortDurations(out), outOfOrder -} - -// batchRows parses " " lines, keeping those whose name carries prefix. An -// unset field renders as "", which simply never matches a caller's wanted -// value. +// The placeholder is collapsed FIRST, and that one substitution is load-bearing: +// go-template renders an unset field as "", which is TWO space-separated tokens, +// so such a row has three fields and the count check below silently DROPS it. That is not +// "the value never matches" — an unscheduled Pod has no .spec.nodeName, so it was missing +// from the parse entirely and the "created" stage could not stamp it until the scheduler +// had already bound it, which is why creation and binding used to report identical +// percentiles. One token keeps the row, with a value no caller looks for. func batchRows(out, prefix string) map[string]string { rows := map[string]string{} - for _, line := range utils.GetNonEmptyLines(out) { + for _, line := range utils.GetNonEmptyLines(strings.ReplaceAll(out, "", "-")) { fields := strings.Fields(line) if len(fields) != 2 || !strings.HasPrefix(fields[0], prefix) { continue @@ -520,6 +545,14 @@ func batchRows(out, prefix string) map[string]string { return rows } +// podRow splits the composite value the Pods poll asks for into the node the Pod is +// bound to and its Ready condition. Either half can be empty — an unscheduled Pod has +// no nodeName (rendered "-", see batchRows) and a fresh one has no conditions. +func podRow(v string) (node, ready string) { + node, ready, _ = strings.Cut(v, "|") + return node, ready +} + // firstSeen keeps the earliest observation of a name; later polls are ignored. func firstSeen(seen map[string]time.Duration, name string, at time.Duration) { if _, dup := seen[name]; !dup { @@ -567,11 +600,16 @@ func reportBatchSync(n int, s syncSamples) { perfDeployName, n) // Not "by the ReplicaSet": that is only true while the batch is a Deployment, and the // stage means the same thing for any workload shape. + // Pods first, then the claim ledger, and each group's derived per-object row last: it is + // computed from the cumulative ones above it, and unlike them it does not move with how + // fast the replicas were created. Ready ends the Pod group because it is the one the + // others are only a step toward — this is when the workloads were usable. _, _ = fmt.Fprintf(GinkgoWriter, " Pods created %s\n", stageLine(s.created, n)) - _, _ = fmt.Fprintf(GinkgoWriter, " NodeClaims Bound %s\n", stageLine(s.claims, n)) _, _ = fmt.Fprintf(GinkgoWriter, " Pods bound to %-22s %s\n", fakeVirtualNode, stageLine(s.bound, n)) - // The one column that does not move with how fast the replicas were created. - _, _ = fmt.Fprintf(GinkgoWriter, " per-Pod sync (created → bound) %s\n", spreadLine(s.sync)) + _, _ = fmt.Fprintf(GinkgoWriter, " Pods Ready %s\n", stageLine(s.ready, n)) + _, _ = fmt.Fprintf(GinkgoWriter, " placement (Pod created → bound) %s\n", spreadLine(s.placement)) + _, _ = fmt.Fprintf(GinkgoWriter, " NodeClaims Bound %s\n", stageLine(s.claims, n)) + _, _ = fmt.Fprintf(GinkgoWriter, " provisioning (NC created → Bound) %s\n", spreadLine(s.provision)) if len(s.bound) == n && s.bound[n-1] > 0 { _, _ = fmt.Fprintf(GinkgoWriter, " throughput %.1f workloads/s\n", float64(n)/s.bound[n-1].Seconds()) @@ -590,8 +628,6 @@ func reportDrain(total, drainTotal time.Duration, d drainSamples) { stageLine(d.podsGone, d.podsKnown)) _, _ = fmt.Fprintf(GinkgoWriter, " NodeClaims gone %s\n", stageLine(d.gone, d.known)) - // The one column here that is not dominated by how fast the virtual kubelet deletes Pods. - _, _ = fmt.Fprintf(GinkgoWriter, " per-claim release (pod → claim gone) %s\n", releaseLine(d)) if len(d.gone) > 0 && drainTotal > 0 { _, _ = fmt.Fprintf(GinkgoWriter, " drain rate %.1f claims/s\n", float64(len(d.gone))/drainTotal.Seconds()) @@ -600,16 +636,6 @@ func reportDrain(total, drainTotal time.Duration, d drainSamples) { _, _ = fmt.Fprintf(GinkgoWriter, " total (apply → all claims gone) %s\n", fmtDuration(total)) } -// releaseLine is the per-claim spread plus the pairs left out of it, so a run never presents -// the distribution without saying what is missing from it. -func releaseLine(d drainSamples) string { - line := fmt.Sprintf("%d/%d %s", len(d.release), d.podsKnown, spreadLine(d.release)) - if d.releaseOutOfOrder > 0 { - line += fmt.Sprintf(" (%d dropped: claim seen gone before its Pod)", d.releaseOutOfOrder) - } - return line -} - // stageLine formats one stage: how many got there, and the spread of when. func stageLine(sorted []time.Duration, n int) string { if len(sorted) == 0 {