diff --git a/.gitignore b/.gitignore index 6ad4836..ebd3370 100644 --- a/.gitignore +++ b/.gitignore @@ -47,3 +47,9 @@ hermes-sandbox-knowledge.md /sandbox-loadgen /sandbox-apiserver /sandbox-sdk-loadgen +/e2ebench +/l2bench +/l3bench +/poolbench +/scalebench +/scalestress diff --git a/api/v1alpha1/sandbox_conversion.go b/api/v1alpha1/sandbox_conversion.go index b3805c4..a40fa9e 100644 --- a/api/v1alpha1/sandbox_conversion.go +++ b/api/v1alpha1/sandbox_conversion.go @@ -25,32 +25,34 @@ import ( const v1alpha1SandboxStateAnnotation = "api.agents.x-k8s.io/v1alpha1-sandbox-state" +// v1alpha1State is the round-trip payload: the fields v1beta1 cannot represent +// (replica counts collapse into OperatingMode). Legacy annotations carrying a +// full v1alpha1 Sandbox JSON decode into this shape too. +type v1alpha1State struct { + Spec struct { + Replicas *int32 `json:"replicas,omitempty"` + } `json:"spec"` + Status struct { + Replicas int32 `json:"replicas,omitempty"` + } `json:"status"` +} + // ConvertTo converts this Sandbox to the Hub version (v1beta1). func (s *Sandbox) ConvertTo(dstRaw conversion.Hub) error { dst := dstRaw.(*v1beta1.Sandbox) - // Copy object metadata s.ObjectMeta.DeepCopyInto(&dst.ObjectMeta) + ConvertSpecTo(&s.Spec, &dst.Spec) + ConvertStatusTo(&s.Status, &dst.Status) - // Convert Spec - if err := ConvertSpecTo(&s.Spec, &dst.Spec); err != nil { - return err - } - - // Convert Status - if err := ConvertStatusTo(&s.Status, &dst.Status); err != nil { - return err - } - - // Preserve the original v1alpha1 object state for lossless round-tripping + // Preserve the fields v1beta1 cannot represent for lossless round-tripping if dst.Annotations == nil { dst.Annotations = make(map[string]string) } - sCopy := s.DeepCopy() - if sCopy.Annotations != nil { - delete(sCopy.Annotations, v1alpha1SandboxStateAnnotation) - } - stateJSON, err := json.Marshal(sCopy) + var state v1alpha1State + state.Spec.Replicas = s.Spec.Replicas + state.Status.Replicas = s.Status.Replicas + stateJSON, err := json.Marshal(state) if err != nil { return fmt.Errorf("failed to marshal v1alpha1 Sandbox state: %w", err) } @@ -63,18 +65,9 @@ func (s *Sandbox) ConvertTo(dstRaw conversion.Hub) error { func (s *Sandbox) ConvertFrom(srcRaw conversion.Hub) error { src := srcRaw.(*v1beta1.Sandbox) - // Copy object metadata src.ObjectMeta.DeepCopyInto(&s.ObjectMeta) - - // Convert Spec - if err := ConvertSpecFrom(&src.Spec, &s.Spec); err != nil { - return err - } - - // Convert Status - if err := ConvertStatusFrom(&src.Status, &s.Status); err != nil { - return err - } + ConvertSpecFrom(&src.Spec, &s.Spec) + ConvertStatusFrom(&src.Status, &s.Status) // Set best-effort default for Status.Replicas based on OperatingMode. // This will be overridden by the restoration logic if the annotation exists. @@ -89,7 +82,7 @@ func (s *Sandbox) ConvertFrom(srcRaw conversion.Hub) error { // Strip the state annotation so it doesn't leak to clients and get sent back on updates delete(s.Annotations, v1alpha1SandboxStateAnnotation) - var original Sandbox + var original v1alpha1State if err := json.Unmarshal([]byte(stateJSON), &original); err != nil { return fmt.Errorf("failed to unmarshal v1alpha1 Sandbox state: %w", err) } @@ -97,48 +90,34 @@ func (s *Sandbox) ConvertFrom(srcRaw conversion.Hub) error { // Restore replicas field from original if OperatingMode matches original intent switch src.Spec.OperatingMode { case v1beta1.SandboxOperatingModeSuspended: - zero := int32(0) - s.Spec.Replicas = &zero + s.Spec.Replicas = new(int32(0)) case v1beta1.SandboxOperatingModeRunning: if original.Spec.Replicas == nil || *original.Spec.Replicas != 0 { s.Spec.Replicas = original.Spec.Replicas } else { - one := int32(1) - s.Spec.Replicas = &one + s.Spec.Replicas = new(int32(1)) } } - // Restore Status replicas s.Status.Replicas = original.Status.Replicas } return nil } -// Helper functions for Sandbox conversion +func ConvertSpecTo(src *SandboxSpec, dst *v1beta1.SandboxSpec) { + ConvertPodTemplateTo(&src.PodTemplate, &dst.PodTemplate) -func ConvertSpecTo(src *SandboxSpec, dst *v1beta1.SandboxSpec) error { - // PodTemplate - if err := ConvertPodTemplateTo(&src.PodTemplate, &dst.PodTemplate); err != nil { - return err - } - - // VolumeClaimTemplates if src.VolumeClaimTemplates != nil { dst.VolumeClaimTemplates = make([]v1beta1.PersistentVolumeClaimTemplate, len(src.VolumeClaimTemplates)) for i := range src.VolumeClaimTemplates { - if err := ConvertPVCClaimTemplateTo(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]); err != nil { - return err - } + ConvertPVCClaimTemplateTo(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]) } } else { dst.VolumeClaimTemplates = nil } - // Lifecycle - if err := ConvertLifecycleTo(&src.Lifecycle, &dst.Lifecycle); err != nil { - return err - } + ConvertLifecycleTo(&src.Lifecycle, &dst.Lifecycle) // Replicas -> OperatingMode if src.Replicas != nil && *src.Replicas == 0 { @@ -147,79 +126,58 @@ func ConvertSpecTo(src *SandboxSpec, dst *v1beta1.SandboxSpec) error { dst.OperatingMode = v1beta1.SandboxOperatingModeRunning } - // Service dst.Service = src.Service - - return nil } -func ConvertSpecFrom(src *v1beta1.SandboxSpec, dst *SandboxSpec) error { - // PodTemplate - if err := ConvertPodTemplateFrom(&src.PodTemplate, &dst.PodTemplate); err != nil { - return err - } +func ConvertSpecFrom(src *v1beta1.SandboxSpec, dst *SandboxSpec) { + ConvertPodTemplateFrom(&src.PodTemplate, &dst.PodTemplate) - // VolumeClaimTemplates if src.VolumeClaimTemplates != nil { dst.VolumeClaimTemplates = make([]PersistentVolumeClaimTemplate, len(src.VolumeClaimTemplates)) for i := range src.VolumeClaimTemplates { - if err := ConvertPVCClaimTemplateFrom(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]); err != nil { - return err - } + ConvertPVCClaimTemplateFrom(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]) } } else { dst.VolumeClaimTemplates = nil } - // Lifecycle - if err := ConvertLifecycleFrom(&src.Lifecycle, &dst.Lifecycle); err != nil { - return err - } + ConvertLifecycleFrom(&src.Lifecycle, &dst.Lifecycle) // OperatingMode -> Replicas - one := int32(1) - zero := int32(0) if src.OperatingMode == v1beta1.SandboxOperatingModeSuspended { - dst.Replicas = &zero + dst.Replicas = new(int32(0)) } else { - dst.Replicas = &one + dst.Replicas = new(int32(1)) } - // Service dst.Service = src.Service - - return nil } -func ConvertStatusTo(src *SandboxStatus, dst *v1beta1.SandboxStatus) error { +func ConvertStatusTo(src *SandboxStatus, dst *v1beta1.SandboxStatus) { dst.ServiceFQDN = src.ServiceFQDN dst.Service = src.Service dst.Conditions = src.Conditions dst.LabelSelector = src.LabelSelector dst.PodIPs = src.PodIPs dst.NodeName = "" // NodeName is new in v1beta1 and does not exist in v1alpha1 - return nil } -func ConvertStatusFrom(src *v1beta1.SandboxStatus, dst *SandboxStatus) error { +func ConvertStatusFrom(src *v1beta1.SandboxStatus, dst *SandboxStatus) { dst.ServiceFQDN = src.ServiceFQDN dst.Service = src.Service dst.Conditions = src.Conditions dst.LabelSelector = src.LabelSelector dst.PodIPs = src.PodIPs - return nil } -func ConvertPodTemplateTo(src *PodTemplate, dst *v1beta1.PodTemplate) error { +func ConvertPodTemplateTo(src *PodTemplate, dst *v1beta1.PodTemplate) { dst.Spec = src.Spec ConvertPodMetadataTo(&src.ObjectMeta, &dst.ObjectMeta) - return nil } -func ConvertPodTemplateFrom(src *v1beta1.PodTemplate, dst *PodTemplate) error { +func ConvertPodTemplateFrom(src *v1beta1.PodTemplate, dst *PodTemplate) { dst.Spec = src.Spec ConvertPodMetadataFrom(&src.ObjectMeta, &dst.ObjectMeta) - return nil } func ConvertPodMetadataTo(src *PodMetadata, dst *v1beta1.PodMetadata) { @@ -232,16 +190,14 @@ func ConvertPodMetadataFrom(src *v1beta1.PodMetadata, dst *PodMetadata) { dst.Annotations = src.Annotations } -func ConvertPVCClaimTemplateTo(src *PersistentVolumeClaimTemplate, dst *v1beta1.PersistentVolumeClaimTemplate) error { +func ConvertPVCClaimTemplateTo(src *PersistentVolumeClaimTemplate, dst *v1beta1.PersistentVolumeClaimTemplate) { dst.Spec = src.Spec ConvertEmbeddedMetadataTo(&src.EmbeddedObjectMetadata, &dst.EmbeddedObjectMetadata) - return nil } -func ConvertPVCClaimTemplateFrom(src *v1beta1.PersistentVolumeClaimTemplate, dst *PersistentVolumeClaimTemplate) error { +func ConvertPVCClaimTemplateFrom(src *v1beta1.PersistentVolumeClaimTemplate, dst *PersistentVolumeClaimTemplate) { dst.Spec = src.Spec ConvertEmbeddedMetadataFrom(&src.EmbeddedObjectMetadata, &dst.EmbeddedObjectMetadata) - return nil } func ConvertEmbeddedMetadataTo(src *EmbeddedObjectMetadata, dst *v1beta1.EmbeddedObjectMetadata) { @@ -256,24 +212,20 @@ func ConvertEmbeddedMetadataFrom(src *v1beta1.EmbeddedObjectMetadata, dst *Embed dst.Annotations = src.Annotations } -func ConvertLifecycleTo(src *Lifecycle, dst *v1beta1.Lifecycle) error { +func ConvertLifecycleTo(src *Lifecycle, dst *v1beta1.Lifecycle) { dst.ShutdownTime = src.ShutdownTime if src.ShutdownPolicy != nil { - policy := v1beta1.ShutdownPolicy(*src.ShutdownPolicy) - dst.ShutdownPolicy = &policy + dst.ShutdownPolicy = new(v1beta1.ShutdownPolicy(*src.ShutdownPolicy)) } else { dst.ShutdownPolicy = nil } - return nil } -func ConvertLifecycleFrom(src *v1beta1.Lifecycle, dst *Lifecycle) error { +func ConvertLifecycleFrom(src *v1beta1.Lifecycle, dst *Lifecycle) { dst.ShutdownTime = src.ShutdownTime if src.ShutdownPolicy != nil { - policy := ShutdownPolicy(*src.ShutdownPolicy) - dst.ShutdownPolicy = &policy + dst.ShutdownPolicy = new(ShutdownPolicy(*src.ShutdownPolicy)) } else { dst.ShutdownPolicy = nil } - return nil } diff --git a/api/v1alpha1/sandbox_conversion_bench_test.go b/api/v1alpha1/sandbox_conversion_bench_test.go new file mode 100644 index 0000000..0aa121e --- /dev/null +++ b/api/v1alpha1/sandbox_conversion_bench_test.go @@ -0,0 +1,85 @@ +package v1alpha1 + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + v1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" +) + +// BenchmarkConvertRoundTrip converts a realistically sized v1alpha1 Sandbox to +// the hub and back — the per-object cost every v1alpha1 read/write pays in the +// conversion webhook, N times per LIST. bytes/ann is the round-trip annotation +// the object carries afterwards. +func BenchmarkConvertRoundTrip(b *testing.B) { + src := benchSandbox() + b.ReportAllocs() + var annBytes int + for b.Loop() { + dst := &v1beta1.Sandbox{} + if err := src.ConvertTo(dst); err != nil { + b.Fatalf("convert to: %v", err) + } + annBytes = len(dst.Annotations[v1alpha1SandboxStateAnnotation]) + back := &Sandbox{} + if err := back.ConvertFrom(dst); err != nil { + b.Fatalf("convert from: %v", err) + } + } + b.ReportMetric(float64(annBytes), "bytes/ann") +} + +func benchSandbox() *Sandbox { + replicas := int32(1) + return &Sandbox{ + ObjectMeta: metav1.ObjectMeta{ + Name: "bench-sandbox", + Namespace: "default", + Labels: map[string]string{"app": "agent", "team": "bench"}, + Annotations: map[string]string{ + "prometheus.io/scrape": "true", + }, + }, + Spec: SandboxSpec{ + Replicas: &replicas, + PodTemplate: PodTemplate{ + ObjectMeta: PodMetadata{Labels: map[string]string{"pod": "agent"}}, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: "agent", + Image: "ghcr.io/cocoonstack/sandbox/rt:24.04", + Command: []string{"/bin/agent", "--serve"}, + Env: []corev1.EnvVar{ + {Name: "MODE", Value: "warm"}, + {Name: "REGION", Value: "sg"}, + }, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("2"), + corev1.ResourceMemory: resource.MustParse("8Gi"), + }, + }, + }, + {Name: "sidecar", Image: "ghcr.io/cocoonstack/sandbox/proxy:1.2"}, + }, + }, + }, + }, + Status: SandboxStatus{ + ServiceFQDN: "bench-sandbox.default.svc.cluster.local", + LabelSelector: "sandbox=bench-sandbox", + PodIPs: []string{"10.244.1.17"}, + Replicas: 1, + Conditions: []metav1.Condition{{ + Type: "Ready", + Status: metav1.ConditionTrue, + Reason: "Ready", + LastTransitionTime: metav1.Now(), + }}, + }, + } +} diff --git a/api/v1alpha1/sandbox_conversion_test.go b/api/v1alpha1/sandbox_conversion_test.go index 2493594..a1a2b84 100644 --- a/api/v1alpha1/sandbox_conversion_test.go +++ b/api/v1alpha1/sandbox_conversion_test.go @@ -204,6 +204,39 @@ func TestSandboxConversion(t *testing.T) { } } +// TestConvertFromLegacyFullObjectState pins backward compatibility: objects +// written before the slim round-trip payload carry a full v1alpha1 Sandbox +// JSON under the state annotation, and its replica fields must still restore. +func TestConvertFromLegacyFullObjectState(t *testing.T) { + five := int32(5) + legacy := &Sandbox{ + Spec: SandboxSpec{Replicas: &five}, + Status: SandboxStatus{Replicas: 4}, + } + legacyJSON, err := json.Marshal(legacy) + if err != nil { + t.Fatalf("marshal legacy state: %v", err) + } + hub := &v1beta1.Sandbox{ + ObjectMeta: metav1.ObjectMeta{ + Name: "legacy", + Namespace: "default", + Annotations: map[string]string{v1alpha1SandboxStateAnnotation: string(legacyJSON)}, + }, + Spec: v1beta1.SandboxSpec{OperatingMode: v1beta1.SandboxOperatingModeRunning}, + } + got := &Sandbox{} + if err := got.ConvertFrom(hub); err != nil { + t.Fatalf("convert from: %v", err) + } + if got.Spec.Replicas == nil || *got.Spec.Replicas != 5 { + t.Fatalf("spec.replicas = %v, want 5", got.Spec.Replicas) + } + if got.Status.Replicas != 4 { + t.Fatalf("status.replicas = %d, want 4", got.Status.Replicas) + } +} + func TestSandboxConversionFromHub(t *testing.T) { // Test conversion of a v1beta1 Sandbox created without v1alpha1 state annotation (e.g. created directly via v1beta1 API) tests := []struct { diff --git a/cmd/sandbox-operator/main.go b/cmd/sandbox-operator/main.go index b96b1d1..000a9d1 100644 --- a/cmd/sandbox-operator/main.go +++ b/cmd/sandbox-operator/main.go @@ -380,7 +380,6 @@ func (o *options) setupExtensionControllers(mgr ctrl.Manager, instrumenter asmet if err := (&extensionscontrollers.SandboxTemplateReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - Recorder: mgr.GetEventRecorder("sandboxtemplate-controller"), Tracer: instrumenter, RouterNamespace: o.webhookNamespace, }).SetupWithManager(mgr, o.templateWorkers); err != nil { diff --git a/cmd/sandbox-sdk-loadgen/main.go b/cmd/sandbox-sdk-loadgen/main.go index d0aeecf..1bcb768 100644 --- a/cmd/sandbox-sdk-loadgen/main.go +++ b/cmd/sandbox-sdk-loadgen/main.go @@ -270,7 +270,8 @@ func main() { fmt.Printf("sandbox-sdk-loadgen %s: ns=%s total=%d concurrency=%d cleanup=%v release-timeout=%s image=%s\n", version, o.namespace, o.total, o.concurrency, o.cleanup, o.releaseTimeout, o.image) - summary := run(ctx, cl, &o) + var seq int64 + summary := createBatch(ctx, cl, &o, &seq, o.total) fmt.Println(summary) if summary.leaked > 0 { fmt.Printf("ERROR: %d sandbox(es) leaked — their node claims were left for the TTL reaper. Do NOT scale this run up until the leak is explained.\n", summary.leaked) @@ -327,8 +328,9 @@ func runCycles(ctx context.Context, cl client.Client, o *options) { } } -// createBatch issues exactly n creates at o.concurrency without cleanup, -// drawing names from seq. Failures consume their slot and are not retried. +// createBatch issues exactly n creates at o.concurrency, drawing names from +// seq and recording each outcome (release included under --cleanup). Failures +// consume their slot and are not retried. func createBatch(ctx context.Context, cl client.Client, o *options, seq *int64, n int) runSummary { var s runSummary var issued atomic.Int64 @@ -344,11 +346,8 @@ func createBatch(ctx context.Context, cl client.Client, o *options, seq *int64, return } name := fmt.Sprintf("%s-%d", o.namegen, atomic.AddInt64(seq, 1)) - if _, err := createOne(ctx, cl, o, name); err != nil { - atomic.AddInt64(&s.failed, 1) - } else { - atomic.AddInt64(&s.created, 1) - } + sb, err := createOne(ctx, cl, o, name) + recordOutcome(ctx, cl, o, &s, sb, err) if o.interval > 0 { sleepCtx(ctx, o.interval) } @@ -408,44 +407,6 @@ func sleepCtx(ctx context.Context, d time.Duration) { } } -// run fans out o.concurrency workers. A worker RESERVES a slot from the shared -// issue counter before each create, so across all workers exactly o.total -// creates are issued — failures consume their slot and are not retried (a retry -// would exceed the requested count). -func run(ctx context.Context, cl client.Client, o *options) runSummary { - var s runSummary - var issued atomic.Int64 - start := time.Now() - var wg sync.WaitGroup - for range o.concurrency { - wg.Go(func() { - for { - if ctx.Err() != nil { - return - } - slot := issued.Add(1) - if slot > int64(o.total) { - return - } - name := fmt.Sprintf("%s-%d", o.namegen, slot) - sb, err := createOne(ctx, cl, o, name) - recordOutcome(ctx, cl, o, &s, sb, err) - if o.interval > 0 { - select { - case <-ctx.Done(): - return - case <-time.After(o.interval): - } - } - } - }) - } - wg.Wait() - s.issued = min(issued.Load(), int64(o.total)) - s.elapsed = time.Since(start) - return s -} - // recordOutcome tallies one create attempt and, with --cleanup, its release. A // release that could not be confirmed counts as leaked, never as released. func recordOutcome(ctx context.Context, cl client.Client, o *options, s *runSummary, sb *sandboxv1beta1.Sandbox, err error) { diff --git a/cmd/sandbox-sdk-loadgen/main_test.go b/cmd/sandbox-sdk-loadgen/main_test.go index 2adda65..0e4ca79 100644 --- a/cmd/sandbox-sdk-loadgen/main_test.go +++ b/cmd/sandbox-sdk-loadgen/main_test.go @@ -33,7 +33,7 @@ func TestRunIssuesExactlyTotal(t *testing.T) { cleanup: true, releaseTimeout: 5 * time.Second, } - s := run(t.Context(), cl, o) + s := createBatch(t.Context(), cl, o, new(int64), o.total) if s.issued != 25 { t.Fatalf("issued = %d, want exactly total (25)", s.issued) @@ -75,7 +75,7 @@ func TestConcurrencyClampedToTotal(t *testing.T) { cleanup: true, releaseTimeout: 5 * time.Second, } - s := run(t.Context(), cl, o) + s := createBatch(t.Context(), cl, o, new(int64), o.total) if s.issued != 2 || s.created != 2 { t.Fatalf("issued=%d created=%d, want exactly 2/2", s.issued, s.created) } diff --git a/controllers/sandbox_controller.go b/controllers/sandbox_controller.go index c7aac1d..a627d82 100644 --- a/controllers/sandbox_controller.go +++ b/controllers/sandbox_controller.go @@ -1183,10 +1183,9 @@ func (r *SandboxReconciler) reconcilePVCs(ctx context.Context, sandbox *sandboxv // handles sandbox expiry by deleting child resources and the sandbox itself if needed. func (r *SandboxReconciler) handleSandboxExpiry(ctx context.Context, sandbox *sandboxv1beta1.Sandbox) (bool, error) { - logger := log.FromContext(ctx) var allErrors error - // Delete pod only if owned by this sandbox + // Delete children only if owned by this sandbox podName := resolvePodName(sandbox) pod := &corev1.Pod{} if err := r.Get(ctx, types.NamespacedName{Name: podName, Namespace: sandbox.Namespace}, pod); err != nil { @@ -1194,43 +1193,16 @@ func (r *SandboxReconciler) handleSandboxExpiry(ctx context.Context, sandbox *sa allErrors = errors.Join(allErrors, fmt.Errorf("failed to get pod: %w", err)) } } else { - ownership, controllerRef := checkOwnership(pod, sandbox) - switch ownership { - case resourceOwnedBySandbox: - if err := r.Delete(ctx, pod); err != nil && !k8serrors.IsNotFound(err) { - allErrors = errors.Join(allErrors, fmt.Errorf("failed to delete pod: %w", err)) - } - case resourceUnowned: - logger.Info("Skipping pod deletion during expiry: pod has no controllerRef pointing to this sandbox", - "Pod.Name", pod.Name, "Sandbox.Name", sandbox.Name) - case resourceOwnedByOther: - logger.Info("Skipping pod deletion during expiry: pod is owned by a different controller", - "Pod.Name", pod.Name, "Sandbox.Name", sandbox.Name, - "Owner.Kind", controllerRef.Kind, "Owner.Name", controllerRef.Name, "Owner.UID", controllerRef.UID) - } + allErrors = errors.Join(allErrors, r.deleteExpiredChild(ctx, sandbox, pod, "pod")) } - // Delete service only if owned by this sandbox service := &corev1.Service{} if err := r.Get(ctx, types.NamespacedName{Name: sandbox.Name, Namespace: sandbox.Namespace}, service); err != nil { if !k8serrors.IsNotFound(err) { allErrors = errors.Join(allErrors, fmt.Errorf("failed to get service: %w", err)) } } else { - ownership, controllerRef := checkOwnership(service, sandbox) - switch ownership { - case resourceOwnedBySandbox: - if err := r.Delete(ctx, service); err != nil && !k8serrors.IsNotFound(err) { - allErrors = errors.Join(allErrors, fmt.Errorf("failed to delete service: %w", err)) - } - case resourceUnowned: - logger.Info("Skipping service deletion during expiry: service has no controllerRef pointing to this sandbox", - "Service.Name", service.Name, "Sandbox.Name", sandbox.Name) - case resourceOwnedByOther: - logger.Info("Skipping service deletion during expiry: service is owned by a different controller", - "Service.Name", service.Name, "Sandbox.Name", sandbox.Name, - "Owner.Kind", controllerRef.Kind, "Owner.Name", controllerRef.Name, "Owner.UID", controllerRef.UID) - } + allErrors = errors.Join(allErrors, r.deleteExpiredChild(ctx, sandbox, service, "service")) } if sandbox.Spec.ShutdownPolicy != nil && *sandbox.Spec.ShutdownPolicy == sandboxv1beta1.ShutdownPolicyDelete { @@ -1247,19 +1219,33 @@ func (r *SandboxReconciler) handleSandboxExpiry(ctx context.Context, sandbox *sa // Drop live-resource status while retaining terminal conditions. conditions := sandbox.Status.Conditions sandbox.Status = sandboxv1beta1.SandboxStatus{Conditions: conditions} - // Update status to mark as expired - meta.SetStatusCondition(&sandbox.Status.Conditions, metav1.Condition{ - Type: string(sandboxv1beta1.SandboxConditionReady), - Status: metav1.ConditionFalse, - ObservedGeneration: sandbox.Generation, - Reason: sandboxv1beta1.SandboxReasonExpired, - Message: msgSandboxExpired, - }) + setSandboxExpiredCondition(sandbox) } return false, allErrors } +// deleteExpiredChild deletes one expired-sandbox child resource, but only when +// this sandbox controls it. +func (r *SandboxReconciler) deleteExpiredChild(ctx context.Context, sandbox *sandboxv1beta1.Sandbox, obj client.Object, kind string) error { + logger := log.FromContext(ctx) + ownership, controllerRef := checkOwnership(obj, sandbox) + switch ownership { + case resourceOwnedBySandbox: + if err := r.Delete(ctx, obj); err != nil && !k8serrors.IsNotFound(err) { + return fmt.Errorf("failed to delete %s: %w", kind, err) + } + case resourceUnowned: + logger.Info("Skipping "+kind+" deletion during expiry: no controllerRef pointing to this sandbox", + "Name", obj.GetName(), "Sandbox.Name", sandbox.Name) + case resourceOwnedByOther: + logger.Info("Skipping "+kind+" deletion during expiry: owned by a different controller", + "Name", obj.GetName(), "Sandbox.Name", sandbox.Name, + "Owner.Kind", controllerRef.Kind, "Owner.Name", controllerRef.Name, "Owner.UID", controllerRef.UID) + } + return nil +} + // checks if the sandbox has expired // returns true if expired, false otherwise // if not expired, also returns the duration to requeue after. diff --git a/docs/scaling-design.md b/docs/scaling-design.md index 42e370f..607e98c 100644 --- a/docs/scaling-design.md +++ b/docs/scaling-design.md @@ -115,10 +115,8 @@ immediately. The `SandboxClaim` is marked `Bound` **asynchronously** — the rec follows the action, exactly as kubelet static Pods record to the apiserver after the container is already running. -**Authorization stays central (correctly).** The gateway runs a -`SubjectAccessReview` + `ResourceQuota` check before delivery. Policy is the part -of Kubernetes that *should* stay centralized; only the ownership-transfer -transaction moves to the node. +**Authorization stays central.** The gateway runs a `SubjectAccessReview` +before delivery; only ownership transfer moves to the node. ```go // ClaimGateway is the node-local fast path for warm-pool claims. @@ -126,8 +124,8 @@ transaction moves to the node. // SandboxClaim object is reconciled to Bound asynchronously afterward. type ClaimGateway interface { // Claim transfers ownership of a node-local warm sandbox to the caller, - // returning connection info. It performs the SubjectAccessReview + - // quota check inline; it does NOT block on writing the SandboxClaim. + // returning connection info. It performs SubjectAccessReview inline; + // it does NOT block on writing the SandboxClaim. Claim(ctx context.Context, req ClaimRequest) (Assignment, error) // Release returns a sandbox to the node-local pool (or tears it down). Release(ctx context.Context, assignment Assignment) error @@ -140,7 +138,6 @@ type ClaimGateway interface { |---|---|---| | Gateway crashes after delivery, before recording `Bound` | Orphan binding → audit-only orphan GC + adopt reconciles the record (the VM is never destroyed on pod-level state — see the delete-authorization contract) | No — eventual consistency | | Node has no warm VM | Falls back to the L1 Kubernetes path (create a new Sandbox) | No | -| Quota exceeded | Gateway rejects inline before delivery | No | **Acceptance:** claim p50 sub-millisecond on the sandboxd tier; orphan-binding rate converges to 0 via GC; `kubectl get sandboxclaims` still shows every claim. @@ -309,4 +306,3 @@ apiserver round-trip, sandboxd delivery (0.2–0.7 ms), and informer convergence the real-microVM claim p95 (926 ms, ~7× the p50) is single-node optimistic-concurrency contention under 100 simultaneous claims — exactly the tail L1's per-pool operator sharding is designed to spread across shards and nodes. - diff --git a/extensions/api/v1alpha1/sandboxclaim_conversion.go b/extensions/api/v1alpha1/sandboxclaim_conversion.go index 9bcad7b..05baf97 100644 --- a/extensions/api/v1alpha1/sandboxclaim_conversion.go +++ b/extensions/api/v1alpha1/sandboxclaim_conversion.go @@ -41,18 +41,9 @@ const ( func (s *SandboxClaim) ConvertTo(dstRaw conversion.Hub) error { dst := dstRaw.(*v1beta1.SandboxClaim) - // Copy object metadata s.ObjectMeta.DeepCopyInto(&dst.ObjectMeta) - - // Convert Spec - if err := convertClaimSpecTo(&s.Spec, &dst.Spec, s.Name, s.Status.SandboxStatus.Name); err != nil { - return err - } - - // Convert Status - if err := convertClaimStatusTo(&s.Status, &dst.Status); err != nil { - return err - } + convertClaimSpecTo(&s.Spec, &dst.Spec, s.Name, s.Status.SandboxStatus.Name) + convertClaimStatusTo(&s.Status, &dst.Status) // Restore the v1beta1-only volumeClaimTemplates from its preservation // annotation, so a v1beta1-authored claim keeps its PVCs after a v1alpha1 hop. @@ -67,39 +58,16 @@ func (s *SandboxClaim) ConvertTo(dstRaw conversion.Hub) error { } } - // Preserve the original v1alpha1 object state for lossless round-tripping - if dst.Annotations == nil { - dst.Annotations = make(map[string]string) - } - sCopy := s.DeepCopy() - if sCopy.Annotations != nil { - delete(sCopy.Annotations, v1alpha1SandboxClaimStateAnnotation) - } - stateJSON, err := json.Marshal(sCopy) - if err != nil { - return fmt.Errorf("failed to marshal v1alpha1 SandboxClaim state: %w", err) - } - dst.Annotations[v1alpha1SandboxClaimStateAnnotation] = string(stateJSON) - - return nil + return stashV1alpha1State(dst, v1alpha1SandboxClaimStateAnnotation, "SandboxClaim", s.DeepCopy()) } // ConvertFrom converts from the Hub version (v1beta1) to this SandboxClaim. func (s *SandboxClaim) ConvertFrom(srcRaw conversion.Hub) error { src := srcRaw.(*v1beta1.SandboxClaim) - // Copy object metadata src.ObjectMeta.DeepCopyInto(&s.ObjectMeta) - - // Convert Spec - if err := convertClaimSpecFrom(&src.Spec, &s.Spec); err != nil { - return err - } - - // Convert Status - if err := convertClaimStatusFrom(&src.Status, &s.Status); err != nil { - return err - } + convertClaimSpecFrom(&src.Spec, &s.Spec) + convertClaimStatusFrom(&src.Status, &s.Status) if err := restoreV1alpha1Spec(s, src); err != nil { return err @@ -151,8 +119,7 @@ func restoreV1alpha1Spec(s *SandboxClaim, src *v1beta1.SandboxClaim) error { s.Spec.WarmPool = original.Spec.WarmPool return nil } - policy := WarmPoolPolicy(src.Spec.WarmPoolRef.Name) - s.Spec.WarmPool = &policy + s.Spec.WarmPool = new(WarmPoolPolicy(src.Spec.WarmPoolRef.Name)) return nil } @@ -176,10 +143,7 @@ func stripRandomSuffix(name string) string { return name } -// Helper functions for SandboxClaim conversion - -func convertClaimSpecTo(src *SandboxClaimSpec, dst *v1beta1.SandboxClaimSpec, claimName, sandboxName string) error { - // Lifecycle +func convertClaimSpecTo(src *SandboxClaimSpec, dst *v1beta1.SandboxClaimSpec, claimName, sandboxName string) { if src.Lifecycle != nil { dst.Lifecycle = &v1beta1.Lifecycle{ ShutdownTime: src.Lifecycle.ShutdownTime, @@ -210,12 +174,8 @@ func convertClaimSpecTo(src *SandboxClaimSpec, dst *v1beta1.SandboxClaimSpec, cl } } - // AdditionalPodMetadata - if err := convertPodMetadataToClaim(&src.AdditionalPodMetadata, &dst.AdditionalPodMetadata); err != nil { - return err - } + sandboxv1alpha1.ConvertPodMetadataTo(&src.AdditionalPodMetadata, &dst.AdditionalPodMetadata) - // Env if src.Env != nil { dst.Env = make([]v1beta1.EnvVar, len(src.Env)) for i := range src.Env { @@ -228,12 +188,9 @@ func convertClaimSpecTo(src *SandboxClaimSpec, dst *v1beta1.SandboxClaimSpec, cl } else { dst.Env = nil } - - return nil } -func convertClaimSpecFrom(src *v1beta1.SandboxClaimSpec, dst *SandboxClaimSpec) error { - // Lifecycle +func convertClaimSpecFrom(src *v1beta1.SandboxClaimSpec, dst *SandboxClaimSpec) { if src.Lifecycle != nil { dst.Lifecycle = &Lifecycle{ ShutdownTime: src.Lifecycle.ShutdownTime, @@ -249,22 +206,16 @@ func convertClaimSpecFrom(src *v1beta1.SandboxClaimSpec, dst *SandboxClaimSpec) dst.TemplateRef = SandboxTemplateRef{ Name: templateName, } - policy := WarmPoolPolicyDefault - dst.WarmPool = &policy + dst.WarmPool = new(WarmPoolPolicyDefault) } else { - policy := WarmPoolPolicy(src.WarmPoolRef.Name) - dst.WarmPool = &policy + dst.WarmPool = new(WarmPoolPolicy(src.WarmPoolRef.Name)) dst.TemplateRef = SandboxTemplateRef{ Name: src.WarmPoolRef.Name, } } - // AdditionalPodMetadata - if err := convertPodMetadataFromClaim(&src.AdditionalPodMetadata, &dst.AdditionalPodMetadata); err != nil { - return err - } + sandboxv1alpha1.ConvertPodMetadataFrom(&src.AdditionalPodMetadata, &dst.AdditionalPodMetadata) - // Env if src.Env != nil { dst.Env = make([]EnvVar, len(src.Env)) for i := range src.Env { @@ -277,36 +228,20 @@ func convertClaimSpecFrom(src *v1beta1.SandboxClaimSpec, dst *SandboxClaimSpec) } else { dst.Env = nil } - - return nil } -func convertClaimStatusTo(src *SandboxClaimStatus, dst *v1beta1.SandboxClaimStatus) error { +func convertClaimStatusTo(src *SandboxClaimStatus, dst *v1beta1.SandboxClaimStatus) { dst.Conditions = src.Conditions dst.SandboxStatus = v1beta1.SandboxStatus{ Name: src.SandboxStatus.Name, PodIPs: src.SandboxStatus.PodIPs, } - return nil } -func convertClaimStatusFrom(src *v1beta1.SandboxClaimStatus, dst *SandboxClaimStatus) error { +func convertClaimStatusFrom(src *v1beta1.SandboxClaimStatus, dst *SandboxClaimStatus) { dst.Conditions = src.Conditions dst.SandboxStatus = SandboxStatus{ Name: src.SandboxStatus.Name, PodIPs: src.SandboxStatus.PodIPs, } - return nil -} - -func convertPodMetadataToClaim(src *sandboxv1alpha1.PodMetadata, dst *sandboxv1beta1.PodMetadata) error { - dst.Labels = src.Labels - dst.Annotations = src.Annotations - return nil -} - -func convertPodMetadataFromClaim(src *sandboxv1beta1.PodMetadata, dst *sandboxv1alpha1.PodMetadata) error { - dst.Labels = src.Labels - dst.Annotations = src.Annotations - return nil } diff --git a/extensions/api/v1alpha1/sandboxtemplate_conversion.go b/extensions/api/v1alpha1/sandboxtemplate_conversion.go index 3bb76cb..95f170d 100644 --- a/extensions/api/v1alpha1/sandboxtemplate_conversion.go +++ b/extensions/api/v1alpha1/sandboxtemplate_conversion.go @@ -15,7 +15,6 @@ package v1alpha1 import ( - "encoding/json" "fmt" "sigs.k8s.io/controller-runtime/pkg/conversion" @@ -31,13 +30,8 @@ const v1alpha1SandboxTemplateStateAnnotation = "api.agents.x-k8s.io/v1alpha1-san func (s *SandboxTemplate) ConvertTo(dstRaw conversion.Hub) error { dst := dstRaw.(*v1beta1.SandboxTemplate) - // Copy object metadata s.ObjectMeta.DeepCopyInto(&dst.ObjectMeta) - - // Convert Spec - if err := convertTemplateSpecTo(&s.Spec, &dst.Spec); err != nil { - return fmt.Errorf("convert spec to v1beta1: %w", err) - } + convertTemplateSpecTo(&s.Spec, &dst.Spec) // Restore v1beta1-only VolumeClaimTemplatesPolicy if present in annotations if policy, ok := s.Annotations["api.agents.x-k8s.io/v1beta1-volume-claim-templates-policy"]; ok { @@ -52,34 +46,15 @@ func (s *SandboxTemplate) ConvertTo(dstRaw conversion.Hub) error { } } - // Preserve the original v1alpha1 object state for lossless round-tripping - if dst.Annotations == nil { - dst.Annotations = make(map[string]string) - } - sCopy := s.DeepCopy() - if sCopy.Annotations != nil { - delete(sCopy.Annotations, v1alpha1SandboxTemplateStateAnnotation) - } - stateJSON, err := json.Marshal(sCopy) - if err != nil { - return fmt.Errorf("failed to marshal v1alpha1 SandboxTemplate state: %w", err) - } - dst.Annotations[v1alpha1SandboxTemplateStateAnnotation] = string(stateJSON) - - return nil + return stashV1alpha1State(dst, v1alpha1SandboxTemplateStateAnnotation, "SandboxTemplate", s.DeepCopy()) } // ConvertFrom converts from the Hub version (v1beta1) to this SandboxTemplate. func (s *SandboxTemplate) ConvertFrom(srcRaw conversion.Hub) error { src := srcRaw.(*v1beta1.SandboxTemplate) - // Copy object metadata src.ObjectMeta.DeepCopyInto(&s.ObjectMeta) - - // Convert Spec - if err := convertTemplateSpecFrom(&src.Spec, &s.Spec); err != nil { - return fmt.Errorf("convert spec from v1beta1: %w", err) - } + convertTemplateSpecFrom(&src.Spec, &s.Spec) // Strip the state annotation if present so it doesn't leak to clients and get sent back on updates delete(s.Annotations, v1alpha1SandboxTemplateStateAnnotation) @@ -97,27 +72,18 @@ func (s *SandboxTemplate) ConvertFrom(srcRaw conversion.Hub) error { return nil } -// Helper functions for SandboxTemplate conversion - -func convertTemplateSpecTo(src *SandboxTemplateSpec, dst *v1beta1.SandboxTemplateSpec) error { - // PodTemplate - if err := sandboxv1alpha1.ConvertPodTemplateTo(&src.PodTemplate, &dst.PodTemplate); err != nil { - return fmt.Errorf("convert podTemplate to v1beta1: %w", err) - } +func convertTemplateSpecTo(src *SandboxTemplateSpec, dst *v1beta1.SandboxTemplateSpec) { + sandboxv1alpha1.ConvertPodTemplateTo(&src.PodTemplate, &dst.PodTemplate) - // VolumeClaimTemplates if src.VolumeClaimTemplates != nil { dst.VolumeClaimTemplates = make([]sandboxv1beta1.PersistentVolumeClaimTemplate, len(src.VolumeClaimTemplates)) for i := range src.VolumeClaimTemplates { - if err := sandboxv1alpha1.ConvertPVCClaimTemplateTo(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]); err != nil { - return fmt.Errorf("convert volumeClaimTemplates[%d] to v1beta1: %w", i, err) - } + sandboxv1alpha1.ConvertPVCClaimTemplateTo(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]) } } else { dst.VolumeClaimTemplates = nil } - // NetworkPolicy if src.NetworkPolicy != nil { dst.NetworkPolicy = &v1beta1.NetworkPolicySpec{ Ingress: src.NetworkPolicy.Ingress, @@ -127,37 +93,23 @@ func convertTemplateSpecTo(src *SandboxTemplateSpec, dst *v1beta1.SandboxTemplat dst.NetworkPolicy = nil } - // NetworkPolicyManagement dst.NetworkPolicyManagement = v1beta1.NetworkPolicyManagement(src.NetworkPolicyManagement) - - // EnvVarsInjectionPolicy dst.EnvVarsInjectionPolicy = v1beta1.EnvVarsInjectionPolicy(src.EnvVarsInjectionPolicy) - - // Service dst.Service = src.Service - - return nil } -func convertTemplateSpecFrom(src *v1beta1.SandboxTemplateSpec, dst *SandboxTemplateSpec) error { - // PodTemplate - if err := sandboxv1alpha1.ConvertPodTemplateFrom(&src.PodTemplate, &dst.PodTemplate); err != nil { - return fmt.Errorf("convert podTemplate from v1beta1: %w", err) - } +func convertTemplateSpecFrom(src *v1beta1.SandboxTemplateSpec, dst *SandboxTemplateSpec) { + sandboxv1alpha1.ConvertPodTemplateFrom(&src.PodTemplate, &dst.PodTemplate) - // VolumeClaimTemplates if src.VolumeClaimTemplates != nil { dst.VolumeClaimTemplates = make([]sandboxv1alpha1.PersistentVolumeClaimTemplate, len(src.VolumeClaimTemplates)) for i := range src.VolumeClaimTemplates { - if err := sandboxv1alpha1.ConvertPVCClaimTemplateFrom(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]); err != nil { - return fmt.Errorf("convert volumeClaimTemplates[%d] from v1beta1: %w", i, err) - } + sandboxv1alpha1.ConvertPVCClaimTemplateFrom(&src.VolumeClaimTemplates[i], &dst.VolumeClaimTemplates[i]) } } else { dst.VolumeClaimTemplates = nil } - // NetworkPolicy if src.NetworkPolicy != nil { dst.NetworkPolicy = &NetworkPolicySpec{ Ingress: src.NetworkPolicy.Ingress, @@ -167,14 +119,7 @@ func convertTemplateSpecFrom(src *v1beta1.SandboxTemplateSpec, dst *SandboxTempl dst.NetworkPolicy = nil } - // NetworkPolicyManagement dst.NetworkPolicyManagement = NetworkPolicyManagement(src.NetworkPolicyManagement) - - // EnvVarsInjectionPolicy dst.EnvVarsInjectionPolicy = EnvVarsInjectionPolicy(src.EnvVarsInjectionPolicy) - - // Service dst.Service = src.Service - - return nil } diff --git a/extensions/api/v1alpha1/sandboxwarmpool_conversion.go b/extensions/api/v1alpha1/sandboxwarmpool_conversion.go index aa3f6a9..d4d871b 100644 --- a/extensions/api/v1alpha1/sandboxwarmpool_conversion.go +++ b/extensions/api/v1alpha1/sandboxwarmpool_conversion.go @@ -15,9 +15,6 @@ package v1alpha1 import ( - "encoding/json" - "fmt" - "sigs.k8s.io/controller-runtime/pkg/conversion" v1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" @@ -29,52 +26,20 @@ const v1alpha1SandboxWarmPoolStateAnnotation = "api.agents.x-k8s.io/v1alpha1-san func (s *SandboxWarmPool) ConvertTo(dstRaw conversion.Hub) error { dst := dstRaw.(*v1beta1.SandboxWarmPool) - // Copy object metadata s.ObjectMeta.DeepCopyInto(&dst.ObjectMeta) + convertWarmPoolSpecTo(&s.Spec, &dst.Spec) + convertWarmPoolStatusTo(&s.Status, &dst.Status) - // Convert Spec - if err := convertWarmPoolSpecTo(&s.Spec, &dst.Spec); err != nil { - return err - } - - // Convert Status - if err := convertWarmPoolStatusTo(&s.Status, &dst.Status); err != nil { - return err - } - - // Preserve the original v1alpha1 object state for lossless round-tripping - if dst.Annotations == nil { - dst.Annotations = make(map[string]string) - } - sCopy := s.DeepCopy() - if sCopy.Annotations != nil { - delete(sCopy.Annotations, v1alpha1SandboxWarmPoolStateAnnotation) - } - stateJSON, err := json.Marshal(sCopy) - if err != nil { - return fmt.Errorf("failed to marshal v1alpha1 SandboxWarmPool state: %w", err) - } - dst.Annotations[v1alpha1SandboxWarmPoolStateAnnotation] = string(stateJSON) - - return nil + return stashV1alpha1State(dst, v1alpha1SandboxWarmPoolStateAnnotation, "SandboxWarmPool", s.DeepCopy()) } // ConvertFrom converts from the Hub version (v1beta1) to this SandboxWarmPool. func (s *SandboxWarmPool) ConvertFrom(srcRaw conversion.Hub) error { src := srcRaw.(*v1beta1.SandboxWarmPool) - // Copy object metadata src.ObjectMeta.DeepCopyInto(&s.ObjectMeta) - - // Convert Spec - if err := convertWarmPoolSpecFrom(&src.Spec, &s.Spec); err != nil { - return err - } - - // Convert Status - if err := convertWarmPoolStatusFrom(&src.Status, &s.Status); err != nil { - return err - } + convertWarmPoolSpecFrom(&src.Spec, &s.Spec) + convertWarmPoolStatusFrom(&src.Status, &s.Status) // Strip the state annotation if present so it doesn't leak to clients and get sent back on updates delete(s.Annotations, v1alpha1SandboxWarmPoolStateAnnotation) @@ -82,11 +47,8 @@ func (s *SandboxWarmPool) ConvertFrom(srcRaw conversion.Hub) error { return nil } -// Helper functions for SandboxWarmPool conversion - -func convertWarmPoolSpecTo(src *SandboxWarmPoolSpec, dst *v1beta1.SandboxWarmPoolSpec) error { - replicas := src.Replicas - dst.Replicas = &replicas +func convertWarmPoolSpecTo(src *SandboxWarmPoolSpec, dst *v1beta1.SandboxWarmPoolSpec) { + dst.Replicas = new(src.Replicas) dst.TemplateRef = v1beta1.SandboxTemplateRef{ Name: src.TemplateRef.Name, } @@ -98,11 +60,9 @@ func convertWarmPoolSpecTo(src *SandboxWarmPoolSpec, dst *v1beta1.SandboxWarmPoo } else { dst.UpdateStrategy = nil } - - return nil } -func convertWarmPoolSpecFrom(src *v1beta1.SandboxWarmPoolSpec, dst *SandboxWarmPoolSpec) error { +func convertWarmPoolSpecFrom(src *v1beta1.SandboxWarmPoolSpec, dst *SandboxWarmPoolSpec) { if src.Replicas != nil { dst.Replicas = *src.Replicas } else { @@ -119,20 +79,16 @@ func convertWarmPoolSpecFrom(src *v1beta1.SandboxWarmPoolSpec, dst *SandboxWarmP } else { dst.UpdateStrategy = nil } - - return nil } -func convertWarmPoolStatusTo(src *SandboxWarmPoolStatus, dst *v1beta1.SandboxWarmPoolStatus) error { +func convertWarmPoolStatusTo(src *SandboxWarmPoolStatus, dst *v1beta1.SandboxWarmPoolStatus) { dst.Replicas = src.Replicas dst.ReadyReplicas = src.ReadyReplicas dst.Selector = src.Selector - return nil } -func convertWarmPoolStatusFrom(src *v1beta1.SandboxWarmPoolStatus, dst *SandboxWarmPoolStatus) error { +func convertWarmPoolStatusFrom(src *v1beta1.SandboxWarmPoolStatus, dst *SandboxWarmPoolStatus) { dst.Replicas = src.Replicas dst.ReadyReplicas = src.ReadyReplicas dst.Selector = src.Selector - return nil } diff --git a/extensions/api/v1alpha1/utils.go b/extensions/api/v1alpha1/utils.go new file mode 100644 index 0000000..5331b45 --- /dev/null +++ b/extensions/api/v1alpha1/utils.go @@ -0,0 +1,41 @@ +// Copyright 2026 The Kubernetes Authors. +// +// 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 v1alpha1 + +import ( + "encoding/json" + "fmt" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// stashV1alpha1State stamps sCopy (a deep copy of the source object, minus its +// own stash annotation) onto dst under key for lossless round-tripping. +func stashV1alpha1State(dst metav1.Object, key, kind string, sCopy metav1.Object) error { + if anns := sCopy.GetAnnotations(); anns != nil { + delete(anns, key) + } + stateJSON, err := json.Marshal(sCopy) + if err != nil { + return fmt.Errorf("failed to marshal v1alpha1 %s state: %w", kind, err) + } + anns := dst.GetAnnotations() + if anns == nil { + anns = make(map[string]string) + } + anns[key] = string(stateJSON) + dst.SetAnnotations(anns) + return nil +} diff --git a/extensions/controllers/queue/simple_sandbox_queue.go b/extensions/controllers/queue/simple_sandbox_queue.go index e416bd5..b733ce9 100644 --- a/extensions/controllers/queue/simple_sandbox_queue.go +++ b/extensions/controllers/queue/simple_sandbox_queue.go @@ -25,17 +25,7 @@ type SandboxKey struct { NodeName string } -// SandboxQueue defines the interface for managing a thread-safe, -// highly concurrent queue of adoptable warm pool sandboxes. -type SandboxQueue interface { - Add(namespacedWarmPoolName string, item SandboxKey) - Get(namespacedWarmPoolName string) (SandboxKey, bool) - GetWithStrategy(namespacedWarmPoolName string, pick func([]SandboxKey) (SandboxKey, bool)) (SandboxKey, bool) - RemoveQueue(namespacedWarmPoolName string) - RemoveItem(namespacedWarmPoolName string, item SandboxKey) -} - -// SimpleSandboxQueue implements SandboxQueue using simple synchronized slices. +// SimpleSandboxQueue is a thread-safe queue of adoptable warm pool sandboxes. type SimpleSandboxQueue struct { // queues is a thread-safe dictionary from warm pool name to a synchronizedQueue queues sync.Map diff --git a/extensions/controllers/sandboxclaim_bench_test.go b/extensions/controllers/sandboxclaim_bench_test.go new file mode 100644 index 0000000..85615bf --- /dev/null +++ b/extensions/controllers/sandboxclaim_bench_test.go @@ -0,0 +1,51 @@ +package controllers + +import ( + "fmt" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + extensionsv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" +) + +// A pool mid-operation: most claims already bound, a tail still waiting. +const ( + benchClaimTotal = 2500 + benchClaimUnbound = 50 +) + +func BenchmarkMapWarmPoolToClaims(b *testing.B) { + scheme := newScheme(b) + warmPool := &extensionsv1beta1.SandboxWarmPool{ + ObjectMeta: metav1.ObjectMeta{Name: "bench-pool", Namespace: "default"}, + } + objs := make([]client.Object, 0, benchClaimTotal+1) + objs = append(objs, warmPool) + for i := range benchClaimTotal { + claim := &extensionsv1beta1.SandboxClaim{ + ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("claim-%d", i), Namespace: "default"}, + Spec: extensionsv1beta1.SandboxClaimSpec{WarmPoolRef: extensionsv1beta1.SandboxWarmPoolRef{Name: warmPool.Name}}, + } + if i >= benchClaimUnbound { + claim.Status.SandboxStatus.Name = fmt.Sprintf("sb-%d", i) + } + objs = append(objs, claim) + } + cl := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithIndex(&extensionsv1beta1.SandboxClaim{}, extensionsv1beta1.WarmPoolRefField, warmPoolRefIndexer). + Build() + r := &SandboxClaimReconciler{Client: cl, Scheme: scheme} + + ctx := b.Context() + b.ReportAllocs() + var requests int + for b.Loop() { + requests = len(r.mapWarmPoolToClaims(ctx, warmPool)) + } + b.ReportMetric(float64(requests), "reqs/event") +} diff --git a/extensions/controllers/sandboxclaim_controller.go b/extensions/controllers/sandboxclaim_controller.go index e40a631..c424b1c 100644 --- a/extensions/controllers/sandboxclaim_controller.go +++ b/extensions/controllers/sandboxclaim_controller.go @@ -65,8 +65,6 @@ const ( claimKind = "SandboxClaim" warmPoolKind = "SandboxWarmPool" - ObservabilityAnnotation = "agents.x-k8s.io/controller-first-observed-at" - immediateRequeueDelay = time.Millisecond // adoptionCacheLagRequeueDelay sets the latency floor for a warm-pool claim: the claim is only marked Bound on the pass that observes the adopted Sandbox. adoptionCacheLagRequeueDelay = 5 * time.Millisecond @@ -166,7 +164,7 @@ func (m *triggeredAdoptionMap) Delete(key types.NamespacedName) { type SandboxClaimReconciler struct { client.Client Scheme *runtime.Scheme - WarmSandboxQueue queue.SandboxQueue + WarmSandboxQueue *queue.SimpleSandboxQueue Recorder events.EventRecorder Tracer asmetrics.Instrumenter MaxConcurrentReconciles int @@ -1074,12 +1072,13 @@ func isSandboxReady(sb *v1beta1.Sandbox) bool { } func isRestrictedDomain(domain string) bool { - for _, d := range restrictedDomains { - if domain == d || strings.HasSuffix(domain, "."+d) { - return true - } - } - return false + return domainInList(domain, restrictedDomains) +} + +func domainInList(domain string, list []string) bool { + return slices.ContainsFunc(list, func(d string) bool { + return domain == d || strings.HasSuffix(domain, "."+d) + }) } // validateAdditionalPodMetadata checks claimMeta for invalid domain or label values upfront. @@ -1117,14 +1116,7 @@ func (r *SandboxClaimReconciler) validateAdditionalPodMetadata(claimMeta *v1beta if isLabel { // Strict Allowlist for labels - allowed := false - for _, d := range allowedDomains { - if domain == d || strings.HasSuffix(domain, "."+d) { - allowed = true - break - } - } - if !allowed { + if !domainInList(domain, allowedDomains) { return fmt.Errorf("label domain %q is not in the allowlist", domain) } } else if isRestrictedDomain(domain) { @@ -1416,11 +1408,7 @@ func mergeVolumeClaimTemplates( return nil, fmt.Errorf("%w: cannot override template volume %q", ErrVolumeClaimTemplatesOverrideForbidden, vct.Name) } } - // Simply append claim VCTs to template VCTs - merged := make([]v1beta1.PersistentVolumeClaimTemplate, 0, len(templateVCTs)+len(claimVCTs)) - merged = append(merged, templateVCTs...) - merged = append(merged, claimVCTs...) - return merged, nil + return slices.Concat(templateVCTs, claimVCTs), nil case extensionsv1beta1.VolumeClaimTemplatesPolicyOverrides: // Merge by Name: claim VCT replaces template VCT by name if they match, and new ones are appended. @@ -1755,9 +1743,14 @@ func (r *SandboxClaimReconciler) mapWarmPoolToClaims(ctx context.Context, obj cl log.FromContext(ctx).Error(err, "failed to list SandboxClaims for SandboxWarmPool", "namespace", warmPool.Namespace, "name", warmPool.Name) return nil } - requests := make([]ctrl.Request, 0, len(claims.Items)) + var requests []ctrl.Request for i := range claims.Items { claim := &claims.Items[i] + // Bound claims are driven by their own Sandbox events; pool churn only + // matters to claims still waiting (e.g. WarmPoolNotFound). + if claim.Status.SandboxStatus.Name != "" { + continue + } requests = append(requests, ctrl.Request{NamespacedName: types.NamespacedName{Namespace: claim.Namespace, Name: claim.Name}}) } return requests @@ -1767,16 +1760,7 @@ func (r *SandboxClaimReconciler) mapWarmPoolToClaims(ctx context.Context, obj cl func (r *SandboxClaimReconciler) SetupWithManager(mgr ctrl.Manager, concurrentWorkers int) error { r.MaxConcurrentReconciles = concurrentWorkers - if err := mgr.GetFieldIndexer().IndexField(context.Background(), &extensionsv1beta1.SandboxClaim{}, extensionsv1beta1.WarmPoolRefField, func(rawObj client.Object) []string { - claim, ok := rawObj.(*extensionsv1beta1.SandboxClaim) - if !ok { - return nil - } - if claim.Spec.WarmPoolRef.Name == "" { - return nil - } - return []string{claim.Spec.WarmPoolRef.Name} - }); err != nil { + if err := mgr.GetFieldIndexer().IndexField(context.Background(), &extensionsv1beta1.SandboxClaim{}, extensionsv1beta1.WarmPoolRefField, warmPoolRefIndexer); err != nil { return err } @@ -1796,6 +1780,15 @@ func (r *SandboxClaimReconciler) SetupWithManager(mgr ctrl.Manager, concurrentWo Complete(r) } +// warmPoolRefIndexer indexes SandboxClaims by spec.warmPoolRef.name. +func warmPoolRefIndexer(rawObj client.Object) []string { + claim, ok := rawObj.(*extensionsv1beta1.SandboxClaim) + if !ok || claim.Spec.WarmPoolRef.Name == "" { + return nil + } + return []string{claim.Spec.WarmPoolRef.Name} +} + // cleanupLegacyNetworkPolicy cleans up any deprecated per-claim NetworkPolicies. func (r *SandboxClaimReconciler) cleanupLegacyNetworkPolicy(ctx context.Context, claim *extensionsv1beta1.SandboxClaim) error { logger := log.FromContext(ctx) @@ -1948,7 +1941,7 @@ func hasClaimExpiredCondition(conditions []metav1.Condition) bool { // sandboxEventHandler implements handler.EventHandler for the SandboxClaimReconciler. type sandboxEventHandler struct { - sandboxQueue queue.SandboxQueue + sandboxQueue *queue.SimpleSandboxQueue } func (h *sandboxEventHandler) Create(ctx context.Context, e event.CreateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { @@ -2055,7 +2048,7 @@ func (h *sandboxEventHandler) Delete(ctx context.Context, e event.DeleteEvent, _ } type warmPoolEventHandler struct { - sandboxQueue queue.SandboxQueue + sandboxQueue *queue.SimpleSandboxQueue } func (h *warmPoolEventHandler) Create(_ context.Context, _ event.CreateEvent, _ workqueue.TypedRateLimitingInterface[reconcile.Request]) { diff --git a/extensions/controllers/sandboxclaim_controller_test.go b/extensions/controllers/sandboxclaim_controller_test.go index 0021139..266f23d 100644 --- a/extensions/controllers/sandboxclaim_controller_test.go +++ b/extensions/controllers/sandboxclaim_controller_test.go @@ -1642,7 +1642,7 @@ func TestSandboxClaimTTLCleanupRequiresPersistedExpiredStatus(t *testing.T) { Namespace: "default", UID: "stale-ttl-claim", Annotations: map[string]string{ - ObservabilityAnnotation: time.Now().Format(time.RFC3339Nano), + asmetrics.ObservabilityAnnotation: time.Now().Format(time.RFC3339Nano), }, }, Spec: extensionsv1beta1.SandboxClaimSpec{ @@ -4286,6 +4286,13 @@ func TestMapWarmPoolToClaims(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "claim-other", Namespace: "default"}, Spec: extensionsv1beta1.SandboxClaimSpec{WarmPoolRef: extensionsv1beta1.SandboxWarmPoolRef{Name: "other-warmpool"}}, } + claimBound := &extensionsv1beta1.SandboxClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-bound", Namespace: "default"}, + Spec: extensionsv1beta1.SandboxClaimSpec{WarmPoolRef: extensionsv1beta1.SandboxWarmPoolRef{Name: warmPoolName}}, + Status: extensionsv1beta1.SandboxClaimStatus{ + SandboxStatus: extensionsv1beta1.SandboxStatus{Name: "claim-bound-sandbox"}, + }, + } warmPool := &extensionsv1beta1.SandboxWarmPool{ ObjectMeta: metav1.ObjectMeta{Name: warmPoolName, Namespace: "default"}, @@ -4298,14 +4305,8 @@ func TestMapWarmPoolToClaims(t *testing.T) { // Let's use the WithIndex option on the fake client builder to support the matchingFields query! fakeClientWithIndex := fake.NewClientBuilder(). WithScheme(scheme). - WithObjects(claim1, claim2, claimOther, warmPool). - WithIndex(&extensionsv1beta1.SandboxClaim{}, extensionsv1beta1.WarmPoolRefField, func(obj client.Object) []string { - c := obj.(*extensionsv1beta1.SandboxClaim) - if c.Spec.WarmPoolRef.Name == "" { - return nil - } - return []string{c.Spec.WarmPoolRef.Name} - }). + WithObjects(claim1, claim2, claimOther, claimBound, warmPool). + WithIndex(&extensionsv1beta1.SandboxClaim{}, extensionsv1beta1.WarmPoolRefField, warmPoolRefIndexer). Build() reconciler := &SandboxClaimReconciler{ @@ -4316,7 +4317,7 @@ func TestMapWarmPoolToClaims(t *testing.T) { requests := reconciler.mapWarmPoolToClaims(t.Context(), warmPool) if len(requests) != 2 { - t.Fatalf("expected 2 requests, got %d", len(requests)) + t.Fatalf("expected 2 requests (bound and other-pool claims excluded), got %d", len(requests)) } expectedNames := map[string]bool{"claim-1": true, "claim-2": true} @@ -5229,7 +5230,7 @@ func TestFlushDoesNotResurrectAClearedAnnotation(t *testing.T) { "the staged observability annotation still has to land") } -func newScheme(t *testing.T) *runtime.Scheme { +func newScheme(t testing.TB) *runtime.Scheme { scheme := runtime.NewScheme() if err := sandboxv1beta1.AddToScheme(scheme); err != nil { t.Fatalf("add to scheme: (%v)", err) diff --git a/extensions/controllers/sandboxtemplate_controller.go b/extensions/controllers/sandboxtemplate_controller.go index da6c437..efd2839 100644 --- a/extensions/controllers/sandboxtemplate_controller.go +++ b/extensions/controllers/sandboxtemplate_controller.go @@ -24,7 +24,6 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/tools/events" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" @@ -38,9 +37,8 @@ import ( // SandboxTemplateReconciler reconciles a SandboxTemplate object. type SandboxTemplateReconciler struct { client.Client - Scheme *runtime.Scheme - Recorder events.EventRecorder - Tracer asmetrics.Instrumenter + Scheme *runtime.Scheme + Tracer asmetrics.Instrumenter // RouterNamespace is the namespace the sandbox-router runs in — the same // namespace the operator is installed into. The managed default NetworkPolicy // only admits ingress from that namespace, so it must track the install diff --git a/extensions/controllers/sandboxtemplate_controller_test.go b/extensions/controllers/sandboxtemplate_controller_test.go index 0d1687b..0dd2d74 100644 --- a/extensions/controllers/sandboxtemplate_controller_test.go +++ b/extensions/controllers/sandboxtemplate_controller_test.go @@ -25,7 +25,6 @@ import ( k8errors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/tools/events" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -215,10 +214,9 @@ func TestSandboxTemplateReconcileNetworkPolicy(t *testing.T) { client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tc.existingObjects...).Build() reconciler := &SandboxTemplateReconciler{ - Client: client, - Scheme: scheme, - Recorder: events.NewFakeRecorder(10), - Tracer: asmetrics.NewNoOp(), + Client: client, + Scheme: scheme, + Tracer: asmetrics.NewNoOp(), } req := reconcile.Request{ @@ -294,10 +292,9 @@ func TestSandboxTemplateReconcile_Vulnerability(t *testing.T) { client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(template, unownedNP).Build() reconciler := &SandboxTemplateReconciler{ - Client: client, - Scheme: scheme, - Recorder: events.NewFakeRecorder(10), - Tracer: asmetrics.NewNoOp(), + Client: client, + Scheme: scheme, + Tracer: asmetrics.NewNoOp(), } req := reconcile.Request{ @@ -342,10 +339,9 @@ func TestSandboxTemplateReconcile_Vulnerability(t *testing.T) { client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(template, unownedNP).Build() reconciler := &SandboxTemplateReconciler{ - Client: client, - Scheme: scheme, - Recorder: events.NewFakeRecorder(10), - Tracer: asmetrics.NewNoOp(), + Client: client, + Scheme: scheme, + Tracer: asmetrics.NewNoOp(), } req := reconcile.Request{ diff --git a/extensions/controllers/sandboxwarmpool_bench_test.go b/extensions/controllers/sandboxwarmpool_bench_test.go new file mode 100644 index 0000000..8611f8c --- /dev/null +++ b/extensions/controllers/sandboxwarmpool_bench_test.go @@ -0,0 +1,110 @@ +package controllers + +import ( + "fmt" + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + + sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" + sandboxcontrollers "github.com/cocoonstack/sandbox-operator/controllers" + extensionsv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" + asmetrics "github.com/cocoonstack/sandbox-operator/internal/metrics" +) + +const benchPoolMembers = 2500 + +// BenchmarkWarmPoolReconcileSteady measures one full steady-state Reconcile of a +// pool at the validated scale — the cost every member event used to pay. +func BenchmarkWarmPoolReconcileSteady(b *testing.B) { + r, pool, _ := newBenchPool(b) + req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: pool.Namespace, Name: pool.Name}} + ctx := b.Context() + b.ReportAllocs() + for b.Loop() { + if _, err := r.Reconcile(ctx, req); err != nil { + b.Fatalf("reconcile: %v", err) + } + } +} + +// BenchmarkPoolMemberDeepCopy measures deep-copying the full member list — the +// per-List cost the informer cache pays for this controller unless the read is +// declared copy-free. +func BenchmarkPoolMemberDeepCopy(b *testing.B) { + _, _, members := newBenchPool(b) + b.ReportAllocs() + for b.Loop() { + for i := range members { + _ = members[i].DeepCopy() + } + } +} + +// newBenchPool builds a reconciler over a steady pool: desired == current == +// benchPoolMembers, every member owned, warm-labeled, hash-fresh, and Ready. +func newBenchPool(b *testing.B) (*SandboxWarmPoolReconciler, *extensionsv1beta1.SandboxWarmPool, []sandboxv1beta1.Sandbox) { + scheme := newScheme(b) + template := &extensionsv1beta1.SandboxTemplate{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "bench-tpl"}, + Spec: extensionsv1beta1.SandboxTemplateSpec{ + SandboxBlueprint: sandboxv1beta1.SandboxBlueprint{ + PodTemplate: sandboxv1beta1.PodTemplate{ + Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "c", Image: "ghcr.io/cocoonstack/sandbox/rt:24.04"}}}, + }, + }, + }, + } + blueprintHash, err := computeSandboxBlueprintHash(template) + if err != nil { + b.Fatalf("blueprint hash: %v", err) + } + pool := &extensionsv1beta1.SandboxWarmPool{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "bench-pool", UID: "pool-uid"}, + Spec: extensionsv1beta1.SandboxWarmPoolSpec{ + Replicas: new(int32(benchPoolMembers)), + TemplateRef: extensionsv1beta1.SandboxTemplateRef{Name: template.Name}, + }, + } + poolNameHash := sandboxcontrollers.NameHash(pool.Name) + members := make([]sandboxv1beta1.Sandbox, benchPoolMembers) + objs := make([]runtime.Object, 0, benchPoolMembers+2) + objs = append(objs, pool, template) + for i := range members { + members[i] = sandboxv1beta1.Sandbox{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: fmt.Sprintf("warm-%d", i), + Labels: map[string]string{ + warmPoolSandboxLabel: poolNameHash, + sandboxTemplateRefHash: SandboxTemplateRefHash(template.Name), + sandboxv1beta1.SandboxLaunchTypeLabel: sandboxv1beta1.SandboxLaunchTypeWarm, + sandboxv1beta1.SandboxTemplateHashLabel: blueprintHash, + }, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: extensionsv1beta1.GroupVersion.String(), + Kind: "SandboxWarmPool", + Name: pool.Name, + UID: pool.UID, + Controller: new(true), + }}, + }, + Spec: sandboxv1beta1.SandboxSpec{SandboxBlueprint: *template.Spec.SandboxBlueprint.DeepCopy()}, + Status: sandboxv1beta1.SandboxStatus{ + Conditions: []metav1.Condition{{ + Type: string(sandboxv1beta1.SandboxConditionReady), + Status: metav1.ConditionTrue, + Reason: "Ready", + LastTransitionTime: metav1.Now(), + }}, + }, + } + objs = append(objs, &members[i]) + } + cl := newFakeClient(scheme, objs...) + return &SandboxWarmPoolReconciler{Client: cl, Scheme: scheme, MaxBatchSize: sandboxCreateDeleteMaxBatchSize, Tracer: asmetrics.NewNoOp()}, pool, members +} diff --git a/extensions/controllers/sandboxwarmpool_controller.go b/extensions/controllers/sandboxwarmpool_controller.go index f1668c1..dc18bc2 100644 --- a/extensions/controllers/sandboxwarmpool_controller.go +++ b/extensions/controllers/sandboxwarmpool_controller.go @@ -32,11 +32,14 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" "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/event" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" @@ -106,7 +109,8 @@ func (r *SandboxWarmPoolReconciler) Reconcile(ctx context.Context, req ctrl.Requ oldStatus := warmPool.Status.DeepCopy() // Reconcile the pool (create or delete Sandboxes as needed) - if err := r.reconcilePool(ctx, warmPool); err != nil { + requeueAfter, err := r.reconcilePool(ctx, warmPool) + if err != nil { return ctrl.Result{}, err } @@ -116,18 +120,18 @@ func (r *SandboxWarmPoolReconciler) Reconcile(ctx context.Context, req ctrl.Requ return ctrl.Result{}, err } - return ctrl.Result{}, nil + return ctrl.Result{RequeueAfter: requeueAfter}, nil } // reconcilePool ensures the correct number of pre-allocated sandboxes exist in the pool. -func (r *SandboxWarmPoolReconciler) reconcilePool(ctx context.Context, warmPool *extensionsv1beta1.SandboxWarmPool) error { +func (r *SandboxWarmPoolReconciler) reconcilePool(ctx context.Context, warmPool *extensionsv1beta1.SandboxWarmPool) (time.Duration, error) { logger := log.FromContext(ctx) // In the L3 writable-aggregation design warm capacity lives per-node in // sandboxd, not as Sandbox CRs, and creating CRs would fight the aggregated // node-local claim path. When disabled, report status without any create/delete. if r.DisableSandboxCRManagement { - return r.reconcilePoolStatusOnly(ctx, warmPool) + return 0, r.reconcilePoolStatusOnly(ctx, warmPool) } // Compute hash of the warm pool name for the pool label @@ -139,12 +143,15 @@ func (r *SandboxWarmPoolReconciler) reconcilePool(ctx context.Context, warmPool warmPoolSandboxLabel: poolNameHash, }) + // Copy-free cache read: items are value snapshots sharing cache-owned maps + // and slices, so every member mutation below deep-copies its target first. if err := r.List(ctx, sandboxList, client.InNamespace(warmPool.Namespace), client.MatchingFields{sandboxWarmPoolLabelIndex: poolNameHash}, + client.UnsafeDisableDeepCopy, ); err != nil { logger.Error(err, "Failed to list sandboxes") - return err + return 0, err } // Fetch template and compute hash once to avoid repeated expensive operations, @@ -157,16 +164,23 @@ func (r *SandboxWarmPoolReconciler) reconcilePool(ctx context.Context, warmPool now := time.Now() var healthySandboxes []sandboxv1beta1.Sandbox + var stuckRecheck time.Duration for _, sb := range activeSandboxes { - if !isSandboxReady(&sb) && !sb.CreationTimestamp.IsZero() && now.Sub(sb.CreationTimestamp.Time) > warmPoolReadinessGracePeriod { - logger.Info("Deleting stuck warm pool sandbox", - "sandbox", sb.Name, - "age", now.Sub(sb.CreationTimestamp.Time).Round(time.Second)) - if err := r.Delete(ctx, &sb); err != nil { - logger.Error(err, "Failed to delete stuck sandbox", "sandbox", sb.Name) - allErrors = errors.Join(allErrors, err) + if !isSandboxReady(&sb) && !sb.CreationTimestamp.IsZero() { + age := now.Sub(sb.CreationTimestamp.Time) + if age > warmPoolReadinessGracePeriod { + logger.Info("Deleting stuck warm pool sandbox", + "sandbox", sb.Name, + "age", age.Round(time.Second)) + if err := r.Delete(ctx, &sb); err != nil { + logger.Error(err, "Failed to delete stuck sandbox", "sandbox", sb.Name) + allErrors = errors.Join(allErrors, err) + } + continue } - continue + // Member events are change-filtered, so the grace deadline needs its + // own timer instead of riding on unrelated pool churn. + stuckRecheck = soonerRequeue(stuckRecheck, warmPoolReadinessGracePeriod-age) } healthySandboxes = append(healthySandboxes, sb) } @@ -253,7 +267,7 @@ func (r *SandboxWarmPoolReconciler) reconcilePool(ctx context.Context, warmPool allErrors = errors.Join(allErrors, tmplErr) } - return allErrors + return stuckRecheck, allErrors } // reconcilePoolStatusOnly reports pool status without creating or deleting any @@ -266,9 +280,11 @@ func (r *SandboxWarmPoolReconciler) reconcilePoolStatusOnly(ctx context.Context, labelSelector := labels.SelectorFromSet(labels.Set{warmPoolSandboxLabel: poolNameHash}) sandboxList := &sandboxv1beta1.SandboxList{} + // Copy-free cache read; this branch only counts, never mutates members. if err := r.List(ctx, sandboxList, client.InNamespace(warmPool.Namespace), client.MatchingFields{sandboxWarmPoolLabelIndex: poolNameHash}, + client.UnsafeDisableDeepCopy, ); err != nil { log.FromContext(ctx).Error(err, "Failed to list sandboxes (status-only reconcile)") return err @@ -291,19 +307,15 @@ func (r *SandboxWarmPoolReconciler) adoptSandbox(ctx context.Context, warmPool * if err := controllerutil.SetControllerReference(warmPool, sb, r.Scheme); err != nil { return err } - setWarmLaunchTypeLabelIfNeeded(sb) + setWarmLaunchTypeLabel(sb) return r.Update(ctx, sb) } -func setWarmLaunchTypeLabelIfNeeded(sb *sandboxv1beta1.Sandbox) bool { +func setWarmLaunchTypeLabel(sb *sandboxv1beta1.Sandbox) { if sb.Labels == nil { sb.Labels = make(map[string]string) } - if sb.Labels[sandboxv1beta1.SandboxLaunchTypeLabel] == sandboxv1beta1.SandboxLaunchTypeWarm { - return false - } sb.Labels[sandboxv1beta1.SandboxLaunchTypeLabel] = sandboxv1beta1.SandboxLaunchTypeWarm - return true } // filterActiveSandboxes filters the list of sandboxes, deleting stale ones and adopting orphans. @@ -313,6 +325,10 @@ func (r *SandboxWarmPoolReconciler) filterActiveSandboxes(ctx context.Context, w var allErrors error vettedHashes := make(map[string]bool) + var currentTemplateRefHash string + if template != nil { + currentTemplateRefHash = SandboxTemplateRefHash(template.Name) + } // Determine the update strategy, defaulting to OnReplenish if not specified or unknown. var updateStrategyType extensionsv1beta1.SandboxWarmPoolUpdateStrategyType @@ -346,7 +362,7 @@ func (r *SandboxWarmPoolReconciler) filterActiveSandboxes(ctx context.Context, w } if tmplErr == nil && (updateStrategy == extensionsv1beta1.RecreateSandboxWarmPoolUpdateStrategyType || isOrphan) { - if r.isSandboxStale(ctx, &sb, template, currentSandboxBlueprintHash, vettedHashes) { + if r.isSandboxStale(ctx, &sb, template, currentTemplateRefHash, currentSandboxBlueprintHash, vettedHashes) { logger.Info("Deleting stale sandbox", "sandbox", sb.Name, "isOrphan", isOrphan) if err := r.Delete(ctx, &sb); err != nil { logger.Error(err, "Failed to delete stale sandbox", "sandbox", sb.Name) @@ -356,21 +372,27 @@ func (r *SandboxWarmPoolReconciler) filterActiveSandboxes(ctx context.Context, w } } - if isControlledByPool && setWarmLaunchTypeLabelIfNeeded(&sb) { - if err := r.Update(ctx, &sb); err != nil { + // sb shares cache-owned maps (copy-free List); mutate deep copies only. + if isControlledByPool && sb.Labels[sandboxv1beta1.SandboxLaunchTypeLabel] != sandboxv1beta1.SandboxLaunchTypeWarm { + fresh := sb.DeepCopy() + setWarmLaunchTypeLabel(fresh) + if err := r.Update(ctx, fresh); err != nil { logger.Error(err, "Failed to update sandbox launch type label", "sandbox", sb.Name) allErrors = errors.Join(allErrors, err) continue } + sb = *fresh } if isOrphan { logger.Info("Adopting orphaned sandbox", "sandbox", sb.Name) - if err := r.adoptSandbox(ctx, warmPool, &sb); err != nil { + fresh := sb.DeepCopy() + if err := r.adoptSandbox(ctx, warmPool, fresh); err != nil { logger.Error(err, "Failed to adopt sandbox", "sandbox", sb.Name) allErrors = errors.Join(allErrors, err) continue } + sb = *fresh } activeSandboxes = append(activeSandboxes, sb) @@ -539,13 +561,14 @@ func (r *SandboxWarmPoolReconciler) isSandboxStale( ctx context.Context, sandbox *sandboxv1beta1.Sandbox, template *extensionsv1beta1.SandboxTemplate, + currentTemplateRefHash string, currentSandboxBlueprintHash string, vettedHashes map[string]bool, ) bool { sandboxHash := sandbox.Labels[sandboxv1beta1.SandboxTemplateHashLabel] // If the templateRefHash doesn't match, it's stale. - if sandbox.Labels[sandboxTemplateRefHash] != SandboxTemplateRefHash(template.Name) { + if sandbox.Labels[sandboxTemplateRefHash] != currentTemplateRefHash { return true } @@ -670,7 +693,7 @@ func (r *SandboxWarmPoolReconciler) SetupWithManager(mgr ctrl.Manager, concurren return ctrl.NewControllerManagedBy(mgr). For(&extensionsv1beta1.SandboxWarmPool{}). - Owns(&sandboxv1beta1.Sandbox{}). + Owns(&sandboxv1beta1.Sandbox{}, builder.WithPredicates(predicate.Or(predicate.LabelChangedPredicate{}, poolMemberChangePredicate()))). WithOptions(controller.Options{MaxConcurrentReconciles: concurrentWorkers}). Watches( &extensionsv1beta1.SandboxTemplate{}, @@ -679,6 +702,30 @@ func (r *SandboxWarmPoolReconciler) SetupWithManager(mgr ctrl.Manager, concurren Complete(r) } +// poolMemberChangePredicate passes the non-label member transitions +// reconcilePool reads — ownership, deletion, Ready flips, and generation bumps +// (orphan blueprint re-vetting); LabelChangedPredicate joins it at registration. +// Anything else (PodIPs, other conditions) would re-scan the pool for nothing. +func poolMemberChangePredicate() predicate.Funcs { + return predicate.Funcs{ + UpdateFunc: func(e event.UpdateEvent) bool { + oldSb, okOld := e.ObjectOld.(*sandboxv1beta1.Sandbox) + newSb, okNew := e.ObjectNew.(*sandboxv1beta1.Sandbox) + if !okOld || !okNew { + return true + } + // The reconcile reads only the controller ref, so compare that — + // DeepEqual over the owner slice costs ~200x per member event. + oldRef, newRef := metav1.GetControllerOf(oldSb), metav1.GetControllerOf(newSb) + return oldSb.Generation != newSb.Generation || + oldSb.DeletionTimestamp.IsZero() != newSb.DeletionTimestamp.IsZero() || + (oldRef == nil) != (newRef == nil) || + (oldRef != nil && oldRef.UID != newRef.UID) || + isSandboxReady(oldSb) != isSandboxReady(newSb) + }, + } +} + // findWarmPoolsForTemplate returns a list of reconcile.Requests for all SandboxWarmPools that reference the template. func (r *SandboxWarmPoolReconciler) findWarmPoolsForTemplate(ctx context.Context, obj client.Object) []reconcile.Request { logger := log.FromContext(ctx) diff --git a/extensions/controllers/sandboxwarmpool_controller_test.go b/extensions/controllers/sandboxwarmpool_controller_test.go index 7b323b3..dbcef7e 100644 --- a/extensions/controllers/sandboxwarmpool_controller_test.go +++ b/extensions/controllers/sandboxwarmpool_controller_test.go @@ -140,10 +140,10 @@ func TestReconcilePool(t *testing.T) { ctx := t.Context() - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify final state - count sandboxes with correct warm pool label @@ -276,10 +276,10 @@ func TestReconcilePoolControllerRef(t *testing.T) { ctx := t.Context() - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) list := &sandboxv1beta1.SandboxList{} @@ -365,7 +365,7 @@ func TestPoolLabelValueInIntegration(t *testing.T) { expectedPoolNameHash := sandboxcontrollers.NameHash(poolName) - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) list := &sandboxv1beta1.SandboxList{} @@ -463,7 +463,7 @@ func TestCreatePoolSandboxPropagatesVolumeClaimTemplates(t *testing.T) { MaxBatchSize: sandboxCreateDeleteMaxBatchSize, } - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) list := &sandboxv1beta1.SandboxList{} @@ -559,7 +559,7 @@ func TestCreatePoolSandboxAppliesSecureDefaults(t *testing.T) { MaxBatchSize: sandboxCreateDeleteMaxBatchSize, } - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) list := &sandboxv1beta1.SandboxList{} @@ -672,9 +672,9 @@ func TestReconcilePoolReadyReplicas(t *testing.T) { ctx := t.Context() - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) require.Equal(t, tc.expectedReadyReplicas, warmPool.Status.ReadyReplicas) @@ -766,7 +766,7 @@ func TestReconcilePoolGCStuckSandboxes(t *testing.T) { } ctx := t.Context() - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) // The stuck sandbox should be deleted and replaced @@ -796,7 +796,7 @@ func TestReconcilePoolGCStuckSandboxes(t *testing.T) { } ctx := t.Context() - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Both should be kept (one healthy, one still within grace period) @@ -896,7 +896,7 @@ func TestReconcilePool_TemplateUpdateRollout(t *testing.T) { ctx := t.Context() // Initial reconciliation to create the sandboxes - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Get initial hash label @@ -925,7 +925,7 @@ func TestReconcilePool_TemplateUpdateRollout(t *testing.T) { require.NotEqual(t, initialHash, updatedHash, "Hashes should differ after template update") // Reconcile again to trigger rollout (or lack thereof) - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify state after update @@ -954,7 +954,7 @@ func TestReconcilePool_TemplateUpdateRollout(t *testing.T) { require.NoError(t, err) // Reconcile to trigger replenishment - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify that we have 2 sandboxes: one old (v1) and one new (v2) @@ -1039,7 +1039,7 @@ func TestReconcilePool_TemplateRefUpdate_SameSpec(t *testing.T) { ctx := t.Context() // Initial reconcile - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) sandboxes := &sandboxv1beta1.SandboxList{} @@ -1073,7 +1073,7 @@ func TestReconcilePool_TemplateRefUpdate_SameSpec(t *testing.T) { require.NoError(t, err) // Reconcile again to trigger rollout - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify state after update @@ -1293,7 +1293,7 @@ func TestReconcilePool_TemplateUpdate_DNSPolicy(t *testing.T) { } // Initial reconcile to create sandboxes - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify initial state @@ -1312,7 +1312,7 @@ func TestReconcilePool_TemplateUpdate_DNSPolicy(t *testing.T) { require.NoError(t, err) // Reconcile again, should trigger rollout (deletion and recreation) - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) // Verify that sandboxes now have the updated DNSPolicy @@ -1372,7 +1372,7 @@ func TestIsSandboxStale_OrphanedSandboxVetting(t *testing.T) { Spec: sandboxv1beta1.SandboxSpec{SandboxBlueprint: sandboxv1beta1.SandboxBlueprint{PodTemplate: sandboxv1beta1.PodTemplate{Spec: *spoofedSpec}}}, } - isStaleSpoofed := r.isSandboxStale(ctx, spoofedOrphan, template, currentSandboxBlueprintHash, vettedHashes) + isStaleSpoofed := r.isSandboxStale(ctx, spoofedOrphan, template, SandboxTemplateRefHash(template.Name), currentSandboxBlueprintHash, vettedHashes) require.True(t, isStaleSpoofed, "Orphaned sandbox with spoofed hash but modified PodSpec should be stale") // Case 2: Orphaned sandbox with matching hash label and genuine/fully vetted PodSpec. @@ -1393,7 +1393,7 @@ func TestIsSandboxStale_OrphanedSandboxVetting(t *testing.T) { Spec: sandboxv1beta1.SandboxSpec{SandboxBlueprint: sandboxv1beta1.SandboxBlueprint{PodTemplate: sandboxv1beta1.PodTemplate{Spec: *genuineSpec}}}, } - isStaleGenuine := r.isSandboxStale(ctx, genuineOrphan, template, currentSandboxBlueprintHash, vettedHashes) + isStaleGenuine := r.isSandboxStale(ctx, genuineOrphan, template, SandboxTemplateRefHash(template.Name), currentSandboxBlueprintHash, vettedHashes) require.False(t, isStaleGenuine, "Orphaned sandbox with genuine fully vetted PodSpec should be fresh") } @@ -1562,7 +1562,7 @@ func TestReconcilePool_EvictionOverride(t *testing.T) { EnableWarmPoolEviction: tc.controllerEnable, } - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) list := &sandboxv1beta1.SandboxList{} @@ -1767,7 +1767,7 @@ func TestReconcilePool_TemplateUpdateRecreate(t *testing.T) { ctx := t.Context() // Initial reconcile - err := r.reconcilePool(ctx, warmPool) + _, err := r.reconcilePool(ctx, warmPool) require.NoError(t, err) sandboxes := &sandboxv1beta1.SandboxList{} @@ -1798,7 +1798,7 @@ func TestReconcilePool_TemplateUpdateRecreate(t *testing.T) { } // Recreate strategy should delete stale sandbox and create a fresh one - err = r.reconcilePool(ctx, warmPool) + _, err = r.reconcilePool(ctx, warmPool) require.NoError(t, err) err = r.List(ctx, sandboxes, client.InNamespace(poolNamespace)) diff --git a/pkg/e2bcompat/lifecycle.go b/pkg/e2bcompat/lifecycle.go index 2c67edb..3e297c3 100644 --- a/pkg/e2bcompat/lifecycle.go +++ b/pkg/e2bcompat/lifecycle.go @@ -310,9 +310,6 @@ func (s *Server) nodesWithSandboxes(r *http.Request) ([]string, error) { // empty body leaves the target at its zero value rather than failing, which is // what pause and fork require. func decodeOptional(w http.ResponseWriter, r *http.Request, out any) error { - if r.Body == nil { - return nil - } err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(out) if errors.Is(err, io.EOF) { return nil diff --git a/pkg/e2bcompat/lookup_bench_test.go b/pkg/e2bcompat/lookup_bench_test.go new file mode 100644 index 0000000..a261977 --- /dev/null +++ b/pkg/e2bcompat/lookup_bench_test.go @@ -0,0 +1,50 @@ +package e2bcompat + +import ( + "fmt" + "net/http/httptest" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/cocoonstack/sandbox-operator/pkg/scale" +) + +// BenchmarkLookupByID resolves one sandbox id living on the last node at the +// 200x2000 fleet projection — the per-request cost of every single-sandbox e2b +// verb (get/delete/pause/connect/fork/snapshot/metrics). The requested id uses +// the published DNS-safe form, the slower of the two accepted spellings. +func BenchmarkLookupByID(b *testing.B) { + const nodes, perNode = 200, 2000 + src := scale.NewStaticInventorySource() + for n := range nodes { + name := fmt.Sprintf("node-%03d", n) + entries := make([]scale.InventoryEntry, perNode) + for i := range entries { + entries[i] = scale.InventoryEntry{ + Name: fmt.Sprintf("sandboxes/sb-%s-%d", name, i), + ID: fmt.Sprintf("sb_%s_%d", name, i), + Phase: "Running", + Address: "10.0.0.1:7777", + } + } + src.Put(&scale.NodeInventory{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Node: name, + Address: "10.0.0.1:7777", + Entries: entries, + }) + } + s, err := NewServer(scale.NewScatterGatherStore(src), Options{Namespace: "sandboxes", AllowAnonymous: true}) + if err != nil { + b.Fatalf("NewServer: %v", err) + } + id := publicID(fmt.Sprintf("sb_node-%03d_%d", nodes-1, perNode-1)) + req := httptest.NewRequest("GET", "/sandboxes/"+id, nil) + b.ReportAllocs() + for b.Loop() { + if _, err := s.lookup(req, id); err != nil { + b.Fatalf("lookup: %v", err) + } + } +} diff --git a/pkg/e2bcompat/sandboxid.go b/pkg/e2bcompat/sandboxid.go index 1d3dfd9..dcf028c 100644 --- a/pkg/e2bcompat/sandboxid.go +++ b/pkg/e2bcompat/sandboxid.go @@ -1,6 +1,9 @@ package e2bcompat -import "strings" +import ( + "strings" + "unicode/utf8" +) // A sandboxd claim id is "sb_" + hex (sandboxd pool/claim.go), whose underscore // is not legal in a DNS label. The e2b SDK derives the in-sandbox envd host as @@ -32,19 +35,45 @@ func publicID(claimID string) string { } // matchesID reports whether a live sandbox's claim id is the one a client asked -// for, accepting both the raw claim id and its published DNS-safe rendering so -// an id observed through either surface keeps working. +// for, accepting both the raw claim id and its published DNS-safe rendering. +// The rendering is compared in place: the store sweep calls this per scanned +// entry, and building publicID per candidate would allocate O(fleet) per lookup. func matchesID(claimID, requested string) bool { if claimID == "" || requested == "" { return false } - return claimID == requested || publicID(claimID) == requested + if claimID == requested { + return true + } + if !isASCII(claimID) { + return publicID(claimID) == requested + } + if len(claimID) != len(requested) { + return false + } + for i := range len(claimID) { + c := claimID[i] + if c >= 'A' && c <= 'Z' { + c += 'a' - 'A' + } + if !isDNSSafe(rune(c)) { + c = '-' + } + if requested[i] != c { + return false + } + } + return true } func needsRewrite(s string) bool { return strings.ContainsFunc(s, func(r rune) bool { return !isDNSSafe(r) }) } +func isASCII(s string) bool { + return !strings.ContainsFunc(s, func(r rune) bool { return r >= utf8.RuneSelf }) +} + // isDNSSafe reports whether r is legal inside a DNS label (RFC 1123): lowercase // alphanumerics and the hyphen. func isDNSSafe(r rune) bool { diff --git a/pkg/e2bcompat/server.go b/pkg/e2bcompat/server.go index be40c23..6cf9719 100644 --- a/pkg/e2bcompat/server.go +++ b/pkg/e2bcompat/server.go @@ -20,6 +20,7 @@ package e2bcompat import ( + "context" "crypto/subtle" "encoding/json" "errors" @@ -29,6 +30,7 @@ import ( "time" "github.com/go-logr/logr" + k8serrors "k8s.io/apimachinery/pkg/api/errors" sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" "github.com/cocoonstack/sandbox-operator/pkg/scale" @@ -84,6 +86,13 @@ type Options struct { Log logr.Logger } +// claimIDResolver is the store fast path resolving one sandbox by node-local +// claim id without materializing the fleet; the scatter-gather store +// implements it, and lookup falls back to a List scan for stores that don't. +type claimIDResolver interface { + GetByClaimID(ctx context.Context, namespace string, match func(claimID string) bool) (*sandboxv1beta1.Sandbox, error) +} + // Server translates e2b REST calls onto a scale.SandboxStore. type Server struct { store scale.SandboxStore @@ -312,6 +321,18 @@ func (s *Server) lookup(r *http.Request, id string) (*sandboxv1beta1.Sandbox, er if strings.TrimSpace(id) == "" { return nil, errSandboxNotFound } + if resolver, ok := s.store.(claimIDResolver); ok { + sb, err := resolver.GetByClaimID(r.Context(), s.opts.Namespace, func(claimID string) bool { + return matchesID(claimID, id) + }) + if err != nil { + if k8serrors.IsNotFound(err) { + return nil, errSandboxNotFound + } + return nil, err + } + return sb, nil + } list, err := s.store.List(r.Context(), scale.ListOptions{Namespace: s.opts.Namespace}) if err != nil { return nil, err @@ -341,13 +362,17 @@ func (s *Server) detailFor(sb *sandboxv1beta1.Sandbox) SandboxDetail { if started.IsZero() { started = time.Now() } + state := StateRunning + if sb.Labels[scale.PhaseLabel] == phaseHibernated { + state = StatePaused + } return SandboxDetail{ TemplateID: templateOf(sb), SandboxID: publicID(sb.Annotations[scale.ClaimIDAnnotation]), ClientID: sb.Status.NodeName, StartedAt: started.UTC().Format(time.RFC3339), EndAt: started.Add(DefaultTimeoutSeconds * time.Second).UTC().Format(time.RFC3339), - State: StateRunning, + State: state, EnvdVersion: s.opts.EnvdVersion, EnvdAccessToken: sb.Annotations[tokenAnnotation], Domain: s.opts.Domain, diff --git a/pkg/e2bcompat/server_test.go b/pkg/e2bcompat/server_test.go index ef59b8d..2bf0877 100644 --- a/pkg/e2bcompat/server_test.go +++ b/pkg/e2bcompat/server_test.go @@ -287,6 +287,60 @@ func TestErrorBodyCarriesMessage(t *testing.T) { } } +// TestLookupUsesClaimIDResolver pins the id-keyed fast path over the real +// scatter-gather store: both id spellings resolve without a fleet List, other +// namespaces stay invisible, and a miss is a plain not-found. +func TestLookupUsesClaimIDResolver(t *testing.T) { + src := scale.NewStaticInventorySource() + src.Put(&scale.NodeInventory{ + ObjectMeta: metav1.ObjectMeta{Name: "node-a"}, + Node: "node-a", + Address: "10.0.0.1:7777", + Entries: []scale.InventoryEntry{ + {Name: "sandboxes/sb-1", ID: "sb_0123abcd", Phase: "Running", Address: "10.0.0.1:7777"}, + {Name: "elsewhere/sb-2", ID: "sb_ffff0000", Phase: "Running", Address: "10.0.0.1:7777"}, + }, + }) + s, err := NewServer(scale.NewScatterGatherStore(src), Options{Namespace: "sandboxes", AllowAnonymous: true}) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + req := httptest.NewRequest(http.MethodGet, "/sandboxes/x", nil) + + for _, id := range []string{"sb_0123abcd", publicID("sb_0123abcd")} { + sb, err := s.lookup(req, id) + if err != nil { + t.Fatalf("lookup(%q): %v", id, err) + } + if sb.Name != "sb-1" || sb.Namespace != "sandboxes" { + t.Fatalf("lookup(%q) = %s/%s, want sandboxes/sb-1", id, sb.Namespace, sb.Name) + } + } + if _, err := s.lookup(req, "sb_ffff0000"); err != errSandboxNotFound { + t.Fatalf("cross-namespace id resolved, want errSandboxNotFound, got %v", err) + } + if _, err := s.lookup(req, "sb_missing"); err != errSandboxNotFound { + t.Fatalf("missing id: want errSandboxNotFound, got %v", err) + } +} + +// TestDetailStateFollowsThePhaseLabel pins the e2b state mapping: a Hibernated +// phase reports paused, anything else running. +func TestDetailStateFollowsThePhaseLabel(t *testing.T) { + s, err := NewServer(&fakeStore{}, Options{AllowAnonymous: true}) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + sb := liveSandbox("sb-1", "sb_0123abcd", "node-a", "img", "tok") + if got := s.detailFor(&sb).State; got != StateRunning { + t.Fatalf("running sandbox state = %q, want %q", got, StateRunning) + } + sb.Labels = map[string]string{scale.PhaseLabel: phaseHibernated} + if got := s.detailFor(&sb).State; got != StatePaused { + t.Fatalf("hibernated sandbox state = %q, want %q", got, StatePaused) + } +} + // fakeStore records what the compat layer asked of the store and replays canned // answers, so the tests assert the translation rather than the node behavior. type fakeStore struct { diff --git a/pkg/scale/claimgateway.go b/pkg/scale/claimgateway.go index 2ab4782..4220ccc 100644 --- a/pkg/scale/claimgateway.go +++ b/pkg/scale/claimgateway.go @@ -2,9 +2,7 @@ package scale import "context" -// ClaimRequest is a node-local warm-pool claim. It names the SandboxClaim being -// served and the warm pool to draw from; the gateway authorizes the caller -// (SubjectAccessReview + ResourceQuota) inline before delivery. +// ClaimRequest identifies a node-local warm-pool claim. type ClaimRequest struct { Namespace string ClaimName string @@ -32,11 +30,7 @@ type Assignment struct { // hands over an already-running guest in sub-millisecond time and returns // connection info immediately; the SandboxClaim object is reconciled to Bound // asynchronously afterward. -// -// Authorization stays central and inline (SubjectAccessReview + ResourceQuota): -// policy is the part of Kubernetes that should remain centralized; only the -// ownership-transfer transaction moves to the node. If the node has no warm VM, -// the caller falls back to the L1 Kubernetes path (create a new Sandbox). +// Authorization stays central through SubjectAccessReview. type ClaimGateway interface { // Claim transfers ownership of a node-local warm sandbox to the caller and // returns connection info. It performs the authorization check inline; it diff --git a/pkg/scale/claimgateway_impl.go b/pkg/scale/claimgateway_impl.go index 13b7eb2..c723bea 100644 --- a/pkg/scale/claimgateway_impl.go +++ b/pkg/scale/claimgateway_impl.go @@ -66,12 +66,7 @@ type SandboxdClient interface { Stats(ctx context.Context, id string) (sandboxd.SandboxStats, error) } -// Authorizer runs the inline policy check (SubjectAccessReview + ResourceQuota) -// before a claim is delivered. Policy is the part of Kubernetes that stays -// centralized; only the ownership-transfer transaction moves to the node. A -// non-nil error rejects the claim inline before any delivery and is NOT a -// fallback signal (IsFallback stays false), matching the "quota exceeded → -// reject inline" failure mode in the README. +// Authorizer checks a claim inline before delivery. type Authorizer interface { Authorize(ctx context.Context, req ClaimRequest) error } @@ -90,7 +85,7 @@ type GatewayConfig struct { Node string // Client delivers and releases sandboxes on this node. Client SandboxdClient - // Authorizer runs the inline SubjectAccessReview + quota check. + // Authorizer checks claims inline. Authorizer Authorizer // Recorder durably records Bound asynchronously after delivery. Recorder ClaimRecorder @@ -380,28 +375,16 @@ type SubjectAccessReviewer interface { Create(ctx context.Context, sar *authzv1.SubjectAccessReview, opts metav1.CreateOptions) (*authzv1.SubjectAccessReview, error) } -// QuotaChecker admits a claim against namespace ResourceQuota. Injected so tests -// need no live quota tracker; nil skips the quota gate. -type QuotaChecker interface { - Admit(ctx context.Context, namespace string) error -} - -// ReviewAuthorizer is the default Authorizer. It authorizes a claim with a central -// SubjectAccessReview against the SandboxClaim resource — the policy check the L2 -// design deliberately keeps on the apiserver — and then an optional ResourceQuota -// admission. Both dependencies are injected, so it exercises the real decision -// path without a live cluster. +// ReviewAuthorizer checks SandboxClaim access through SubjectAccessReview. type ReviewAuthorizer struct { Reviewer SubjectAccessReviewer - Quota QuotaChecker // Group/Resource/Verb default to the SandboxClaim create check when empty. Group string Resource string Verb string } -// Authorize denies fail-closed when the caller identity is absent, then runs the -// SubjectAccessReview and (if configured) the quota check. +// Authorize denies missing identities and runs SubjectAccessReview. func (a *ReviewAuthorizer) Authorize(ctx context.Context, req ClaimRequest) error { user := req.Selector[RequestUserSelectorKey] if user == "" { @@ -426,10 +409,5 @@ func (a *ReviewAuthorizer) Authorize(ctx context.Context, req ClaimRequest) erro if got.Status.Denied || !got.Status.Allowed { return fmt.Errorf("user %q may not claim in %s: %s", user, req.Namespace, got.Status.Reason) } - if a.Quota != nil { - if err := a.Quota.Admit(ctx, req.Namespace); err != nil { - return fmt.Errorf("quota: %w", err) - } - } return nil } diff --git a/pkg/scale/claimgateway_impl_test.go b/pkg/scale/claimgateway_impl_test.go index 781f77f..823735d 100644 --- a/pkg/scale/claimgateway_impl_test.go +++ b/pkg/scale/claimgateway_impl_test.go @@ -150,9 +150,7 @@ func TestClaimGatewayReleaseOwnerTeardownOnly(t *testing.T) { require.Equal(t, int64(1), fs.releases.Load()) } -// TestClaimGatewayAuthorizationRejectsInline exercises the default ReviewAuthorizer: -// a denied SubjectAccessReview (or exceeded quota) rejects the claim inline BEFORE -// any delivery, and the rejection is a hard error, not a fallback. +// TestClaimGatewayAuthorizationRejectsInline verifies SAR denials precede delivery. func TestClaimGatewayAuthorizationRejectsInline(t *testing.T) { t.Run("deny", func(t *testing.T) { fs := newFakeSandboxd(t) diff --git a/pkg/scale/sandboxd/lifecycle.go b/pkg/scale/sandboxd/lifecycle.go index e3031b7..6d5f984 100644 --- a/pkg/scale/sandboxd/lifecycle.go +++ b/pkg/scale/sandboxd/lifecycle.go @@ -303,9 +303,6 @@ func (c *Client) getJSON(ctx context.Context, path string, out any) error { // decodeInto reads a bounded reply body into out. func decodeInto(resp *http.Response, out any, path string) error { - if out == nil { - return nil - } b, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) if err != nil { return fmt.Errorf("sandboxd: read %s reply: %w", path, err) diff --git a/pkg/scale/sandboxstore_bench_test.go b/pkg/scale/sandboxstore_bench_test.go new file mode 100644 index 0000000..262a346 --- /dev/null +++ b/pkg/scale/sandboxstore_bench_test.go @@ -0,0 +1,91 @@ +package scale + +import ( + "fmt" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" +) + +// Fleet shapes: the deployed 26-node fleet and the 50k-microVM projection. +var benchFleets = []struct { + name string + nodes int + perNode int +}{ + {"26x100", 26, 100}, + {"200x2000", 200, 2000}, +} + +// BenchmarkStoreGet resolves one sandbox living on the last node in enumeration +// order — the sequential worst case a Get pays on a miss-heavy sweep. +func BenchmarkStoreGet(b *testing.B) { + for _, fleet := range benchFleets { + b.Run(fleet.name, func(b *testing.B) { + store, _ := benchStore(b, fleet.nodes, fleet.perNode) + last := benchNodeName(fleet.nodes - 1) + target := fmt.Sprintf("default/sb-%s-%d", last, fleet.perNode-1) + ns, name := splitNamespacedName(target) + ctx := b.Context() + b.ReportAllocs() + for b.Loop() { + if _, err := store.Get(ctx, ns, name); err != nil { + b.Fatalf("get: %v", err) + } + } + }) + } +} + +// BenchmarkStoreWarmCandidates sweeps every node inventory for warm capacity — +// the node-pick cost every aggregated Create/claim pays. +func BenchmarkStoreWarmCandidates(b *testing.B) { + for _, fleet := range benchFleets { + b.Run(fleet.name, func(b *testing.B) { + store, pool := benchStore(b, fleet.nodes, fleet.perNode) + ctx := b.Context() + b.ReportAllocs() + for b.Loop() { + candidates, err := store.warmCandidates(ctx, pool) + if err != nil { + b.Fatalf("warm candidates: %v", err) + } + if len(candidates) != fleet.nodes { + b.Fatalf("got %d candidates, want %d", len(candidates), fleet.nodes) + } + } + }) + } +} + +// benchStore publishes nodes×perNode inventory entries, every node advertising +// warm capacity for the returned pool key. +func benchStore(b *testing.B, nodes, perNode int) (*scatterGatherStore, PoolKey) { + b.Helper() + pool := PoolKey{Template: "ghcr.io/cocoonstack/sandbox/rt:24.04", Net: NetDefault, Size: SizeClassSmall} + src := NewStaticInventorySource() + for n := range nodes { + name := benchNodeName(n) + entries := make([]InventoryEntry, perNode) + for i := range entries { + entries[i] = InventoryEntry{ + Name: fmt.Sprintf("default/sb-%s-%d", name, i), + ID: fmt.Sprintf("sb_%s_%d", name, i), + Phase: "Running", + Address: "10.0.0.1:7777", + } + } + src.Put(&NodeInventory{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Node: name, + Address: "10.0.0.1:7777", + Entries: entries, + Pools: []extv1beta1.PoolCapacity{{Template: pool.Template, Net: pool.Net, Size: pool.Size, Warm: 5, Target: 5}}, + }) + } + return NewScatterGatherStore(src), pool +} + +func benchNodeName(n int) string { return fmt.Sprintf("node-%03d", n) } diff --git a/pkg/scale/sandboxstore_impl.go b/pkg/scale/sandboxstore_impl.go index bf41534..eb4621e 100644 --- a/pkg/scale/sandboxstore_impl.go +++ b/pkg/scale/sandboxstore_impl.go @@ -114,11 +114,6 @@ func WithLogger(log logr.Logger) StoreOption { return func(s *scatterGatherStore) { s.log = log } } -// WithConcurrency bounds the per-node fan-out. Values <=0 leave it unbounded. -func WithConcurrency(n int) StoreOption { - return func(s *scatterGatherStore) { s.concurrency = n } -} - // WithWatchPollInterval sets how often Watch re-derives node inventory to emit // deltas. Defaults to one second. func WithWatchPollInterval(d time.Duration) StoreOption { @@ -207,77 +202,144 @@ func (s *scatterGatherStore) List(ctx context.Context, opts ListOptions) (*sandb if err != nil { return nil, err } + items, err := fanOutNodes(ctx, s, func(gctx context.Context, node string) []sandboxv1beta1.Sandbox { + inv, invErr := s.src.NodeInventory(gctx, node) + if invErr != nil { + // Partitioned / lost inventory: skip this node's sandboxes rather + // than failing the list. They reappear on the node's next publish. + s.log.V(1).Info("node inventory unavailable; omitting from list (eventual consistency)", + "node", node, "err", invErr.Error()) + return nil + } + return s.materialize(inv, opts.Namespace, labelSel, fieldSel) + }) + if err != nil { + return nil, err + } + list := &sandboxv1beta1.SandboxList{} + list.SetGroupVersionKind(sandboxv1beta1.GroupVersion.WithKind("SandboxList")) + list.Items = items + if list.Items == nil { + list.Items = []sandboxv1beta1.Sandbox{} // an empty list serializes as [], not null + } + slices.SortFunc(list.Items, func(a, b sandboxv1beta1.Sandbox) int { + return cmp.Or(cmp.Compare(a.Namespace, b.Namespace), cmp.Compare(a.Name, b.Name)) + }) + return list, nil +} + +// fanOutNodes enumerates the node inventories and runs work per node with +// List's bounded concurrency, concatenating per-node results in node order. +// A node the work skips contributes nil. +func fanOutNodes[T any](ctx context.Context, s *scatterGatherStore, work func(ctx context.Context, node string) []T) ([]T, error) { nodes, err := s.src.ListNodes(ctx) if err != nil { return nil, fmt.Errorf("scale: enumerate node inventories: %w", err) } - - perNode := make([][]sandboxv1beta1.Sandbox, len(nodes)) + perNode := make([][]T, len(nodes)) g, gctx := errgroup.WithContext(ctx) if s.concurrency > 0 { g.SetLimit(s.concurrency) } for i, node := range nodes { g.Go(func() error { - inv, err := s.src.NodeInventory(gctx, node) - if err != nil { - // Partitioned / lost inventory: skip this node's sandboxes rather - // than failing the list. They reappear on the node's next publish. - s.log.V(1).Info("node inventory unavailable; omitting from list (eventual consistency)", - "node", node, "err", err.Error()) - return nil - } - perNode[i] = s.materialize(inv, opts.Namespace, labelSel, fieldSel) + perNode[i] = work(gctx, node) return nil }) } - if err := g.Wait(); err != nil { - return nil, fmt.Errorf("scale: fan-out to node inventories: %w", err) - } - - total := 0 - for _, chunk := range perNode { - total += len(chunk) - } - list := &sandboxv1beta1.SandboxList{} - list.SetGroupVersionKind(sandboxv1beta1.GroupVersion.WithKind("SandboxList")) - list.Items = make([]sandboxv1beta1.Sandbox, 0, total) - for _, chunk := range perNode { - list.Items = append(list.Items, chunk...) - } - slices.SortFunc(list.Items, func(a, b sandboxv1beta1.Sandbox) int { - return cmp.Or(cmp.Compare(a.Namespace, b.Namespace), cmp.Compare(a.Name, b.Name)) - }) - return list, nil + _ = g.Wait() + return slices.Concat(perNode...), nil } // Get routes to the owning node's authoritative inventory. It resolves which // node holds namespace/name and returns that entry synthesized as a Sandbox. +// The per-node sweep fans out like List and cancels on the first hit — a +// sandbox lives on exactly one node, so first-found is the answer. // // In production this would RPC the owning node's live sandboxd for a // read-after-write answer; in this substrate the node's published NodeInventory // is the authoritative view available, so Get returns from it directly rather // than from an eventually-consistent cluster-wide summary. func (s *scatterGatherStore) Get(ctx context.Context, namespace, name string) (*sandboxv1beta1.Sandbox, error) { + found, err := s.findEntry(ctx, "get", func(inv *NodeInventory, i int) bool { + ens, ename := splitNamespacedName(inv.Entries[i].Name) + return ens == namespace && ename == name + }) + if err != nil { + return nil, err + } + if found == nil { + return nil, k8serrors.NewNotFound(sandboxv1beta1.Resource("sandboxes"), name) + } + return found, nil +} + +// findEntry sweeps every node inventory with List's bounded fan-out and returns +// the first entry matching match synthesized as a Sandbox, canceling the rest +// of the sweep on the hit. Nil with a nil error means no entry matched. +func (s *scatterGatherStore) findEntry(ctx context.Context, op string, match func(inv *NodeInventory, i int) bool) (*sandboxv1beta1.Sandbox, error) { nodes, err := s.src.ListNodes(ctx) if err != nil { return nil, fmt.Errorf("scale: enumerate node inventories: %w", err) } + gctx, cancel := context.WithCancel(ctx) + defer cancel() + g := &errgroup.Group{} + if s.concurrency > 0 { + g.SetLimit(s.concurrency) + } + var ( + mu sync.Mutex + found *sandboxv1beta1.Sandbox + ) for _, node := range nodes { - inv, err := s.src.NodeInventory(ctx, node) - if err != nil { - s.log.V(1).Info("node inventory unavailable during get; trying next node", - "node", node, "err", err.Error()) - continue - } - for i := range inv.Entries { - ens, ename := splitNamespacedName(inv.Entries[i].Name) - if ens == namespace && ename == name { - return entryToSandbox(inv.Node, inv.Entries[i]), nil + g.Go(func() error { + if gctx.Err() != nil { + return nil + } + inv, err := s.src.NodeInventory(gctx, node) + if err != nil { + s.log.V(1).Info("node inventory unavailable during "+op+"; skipping node", + "node", node, "err", err.Error()) + return nil + } + for i := range inv.Entries { + if !match(inv, i) { + continue + } + mu.Lock() + if found == nil { + found = entryToSandbox(inv.Node, inv.Entries[i]) + } + mu.Unlock() + cancel() + return nil } + return nil + }) + } + _ = g.Wait() + return found, nil +} + +// GetByClaimID resolves the sandbox whose node-local claim id satisfies match, +// fanning out per node and canceling on the first hit; only the matching entry +// is materialized. An empty namespace matches every namespace. +func (s *scatterGatherStore) GetByClaimID(ctx context.Context, namespace string, match func(claimID string) bool) (*sandboxv1beta1.Sandbox, error) { + found, err := s.findEntry(ctx, "claim-id get", func(inv *NodeInventory, i int) bool { + if inv.Entries[i].ID == "" || !match(inv.Entries[i].ID) { + return false } + ns, _ := splitNamespacedName(inv.Entries[i].Name) + return namespace == "" || ns == namespace + }) + if err != nil { + return nil, err + } + if found == nil { + return nil, k8serrors.NewNotFound(sandboxv1beta1.Resource("sandboxes"), "") } - return nil, k8serrors.NewNotFound(sandboxv1beta1.Resource("sandboxes"), name) + return found, nil } // Claim picks the node with the most warm capacity for pool and hands over one of @@ -341,32 +403,28 @@ func (s *scatterGatherStore) Release(ctx context.Context, node, id string) error return nil } -// warmCandidates lists every node advertising warm capacity for pool. A node whose -// inventory is unavailable is skipped, not fatal: the fleet stays claimable while -// one node is partitioned. +// warmCandidates lists every node advertising warm capacity for pool, fanning +// out per node like List. A node whose inventory is unavailable is skipped, not +// fatal: the fleet stays claimable while one node is partitioned. func (s *scatterGatherStore) warmCandidates(ctx context.Context, pool PoolKey) ([]warmCandidate, error) { - nodes, err := s.src.ListNodes(ctx) - if err != nil { - return nil, fmt.Errorf("scale: enumerate node inventories: %w", err) - } - var out []warmCandidate - for _, n := range nodes { - inv, err := s.src.NodeInventory(ctx, n) + return fanOutNodes(ctx, s, func(gctx context.Context, n string) []warmCandidate { + inv, err := s.src.NodeInventory(gctx, n) if err != nil { s.log.V(1).Info("node inventory unavailable during claim node-pick; skipping", "node", n, "err", err.Error()) - continue + return nil } if inv.Address == "" { - continue + return nil } - for i := range inv.Pools { - if pc := inv.Pools[i]; pc.Warm > 0 && poolCapacityMatches(pc, pool) { + var out []warmCandidate + for j := range inv.Pools { + if pc := inv.Pools[j]; pc.Warm > 0 && poolCapacityMatches(pc, pool) { out = append(out, warmCandidate{node: n, addr: inv.Address, warm: pc.Warm}) } } - } - return out, nil + return out + }) } // Watch merges per-node inventory into a single Sandbox event stream. This @@ -703,25 +761,6 @@ func (p *NodeInventoryPublisher) Publish(ctx context.Context) (int, error) { return len(entries), nil } -// PublishPeriodically runs Publish on interval until ctx is canceled. Publish -// failures are logged, not fatal — the next tick rebuilds from live state. -func (p *NodeInventoryPublisher) PublishPeriodically(ctx context.Context, interval time.Duration) { - ticker := time.NewTicker(interval) - defer ticker.Stop() - for { - if n, err := p.Publish(ctx); err != nil { - p.log.Error(err, "node inventory publish failed; will retry on next tick", "node", p.node) - } else { - p.log.V(1).Info("published node inventory", "node", p.node, "entries", n) - } - select { - case <-ctx.Done(): - return - case <-ticker.C: - } - } -} - var _ InventoryApplier = (*ssaInventoryApplier)(nil) // ssaInventoryApplier is the production applier: it server-side-applies the diff --git a/pkg/scale/warmpool/driver.go b/pkg/scale/warmpool/driver.go index 5ca3a8a..9d84231 100644 --- a/pkg/scale/warmpool/driver.go +++ b/pkg/scale/warmpool/driver.go @@ -18,14 +18,16 @@ import ( "errors" "fmt" "slices" - "sync" "time" "github.com/go-logr/logr" + "golang.org/x/sync/errgroup" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" @@ -142,8 +144,11 @@ func (d *Driver) SetupWithManager(mgr ctrl.Manager) error { enqueueAll := handler.EnqueueRequestsFromMapFunc(func(context.Context, client.Object) []reconcile.Request { return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: "sync"}}} }) + // Generation-filtered so the loop's own writeStatus cannot re-trigger it + // into a continuous back-to-back loop under claim churn; create/delete, + // spec edits, NodeInventory events, and the RequeueAfter tick keep coverage. return ctrl.NewControllerManagedBy(mgr). - For(&extv1beta1.SandboxWarmPool{}). + For(&extv1beta1.SandboxWarmPool{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&extv1beta1.NodeInventory{}, enqueueAll). Named("sandboxwarmpool"). Complete(d) @@ -254,8 +259,8 @@ func (d *Driver) applyToNodes(ctx context.Context, nodes []nodeView, desired []d d.log.Info("warm-pool driver has no sandboxd token; skipping pool apply (fail-closed)") return } - sem := make(chan struct{}, maxNodeConcurrency) - var wg sync.WaitGroup + var g errgroup.Group + g.SetLimit(maxNodeConcurrency) for i := range nodes { node := nodes[i] // Aggregate targets BY KEY: two SandboxWarmPools may resolve to the same @@ -276,18 +281,16 @@ func (d *Driver) applyToNodes(ctx context.Context, nodes []nodeView, desired []d cmp.Compare(a.Size, b.Size), ) }) - sem <- struct{}{} - wg.Go(func() { - defer func() { <-sem }() + g.Go(func() error { callCtx, cancel := context.WithTimeout(ctx, setPoolsTimeout) defer cancel() info, err := d.factory(node.addr, d.token).SetPools(callCtx, specs) if err != nil { d.log.Error(err, "set node warm pools", "node", node.name, "addr", node.addr) - return + return nil } if info == nil { - return + return nil } // The PUT response carries this node's live per-pool warm, so adopt it // as the status source: NodeInventory lags by up to one publish @@ -296,9 +299,10 @@ func (d *Driver) applyToNodes(ctx context.Context, nodes []nodeView, desired []d // failed keeps its inventory-derived counts. Only this goroutine // touches nodes[i], so no lock is needed. nodes[i].warmBy = warmByFrom(info) + return nil }) } - wg.Wait() + _ = g.Wait() } // writeStatus updates a SandboxWarmPool's status.replicas/readyReplicas from the diff --git a/pkg/scale/warmpool/driver_test.go b/pkg/scale/warmpool/driver_test.go index 9c15cda..af39e68 100644 --- a/pkg/scale/warmpool/driver_test.go +++ b/pkg/scale/warmpool/driver_test.go @@ -3,6 +3,7 @@ package warmpool import ( "context" "errors" + "fmt" "sync" "testing" @@ -216,6 +217,24 @@ func TestApplyBoundsEachNodeCall(t *testing.T) { } } +// BenchmarkReconcileOnce measures one global driver pass — the cost of every +// wake-up, spurious or not — at the deployed fleet shape (26 nodes, 4 pools). +func BenchmarkReconcileOnce(b *testing.B) { + objs := []client.Object{template()} + for i := range 4 { + objs = append(objs, warmPool(fmt.Sprintf("p%d", i), 500)) + } + d, _, inv, _ := newTestDriver(b, objs...) + putNodes(inv, 26) + ctx := b.Context() + b.ReportAllocs() + for b.Loop() { + if err := d.reconcileOnce(ctx); err != nil { + b.Fatalf("reconcile: %v", err) + } + } +} + // fakeSetter records the last pools set per node address and plays back the warm // counts a node reports in its PUT response — the driver's status source. type fakeSetter struct { @@ -270,7 +289,7 @@ func (n *fakeNode) SetPools(ctx context.Context, pools []sandboxd.PoolSpec) (*sa return info, nil } -func newTestDriver(t *testing.T, objs ...client.Object) (*Driver, *fakeSetter, *scale.StaticInventorySource, client.Client) { +func newTestDriver(t testing.TB, objs ...client.Object) (*Driver, *fakeSetter, *scale.StaticInventorySource, client.Client) { t.Helper() scheme := runtime.NewScheme() if err := extv1beta1.AddToScheme(scheme); err != nil { diff --git a/test/benchutil/benchutil.go b/test/benchutil/benchutil.go new file mode 100644 index 0000000..a2bd1bb --- /dev/null +++ b/test/benchutil/benchutil.go @@ -0,0 +1,106 @@ +// Package benchutil holds the plumbing shared by the test/* bench harnesses: +// fatal-error exit, truncating percentiles, and warm-pool fixture helpers. +package benchutil + +import ( + "context" + "fmt" + "os" + "slices" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" + extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" +) + +// Must exits the harness when err is non-nil. +func Must(err error) { + if err != nil { + fmt.Fprintln(os.Stderr, "fatal:", err) + os.Exit(2) + } +} + +// Pct returns the p-th percentile of xs, truncated by round. +func Pct(xs []float64, p float64, round func(float64) float64) float64 { + if len(xs) == 0 { + return 0 + } + s := slices.Clone(xs) + slices.Sort(s) + if p >= 100 { + return round(s[len(s)-1]) + } + return round(s[int(float64(len(s)-1)*p/100)]) +} + +// Round0..Round4 truncate to the named number of decimals — truncation, not +// rounding, matching the values the harness reports have always emitted. +func Round0(f float64) float64 { return float64(int(f)) } +func Round1(f float64) float64 { return float64(int(f*10)) / 10 } +func Round2(f float64) float64 { return float64(int(f*100)) / 100 } +func Round3(f float64) float64 { return float64(int(f*1000)) / 1000 } +func Round4(f float64) float64 { return float64(int(f*10000)) / 10000 } + +// EnsurePool creates the SandboxWarmPool (labels may be nil) or updates its +// replica count. +func EnsurePool(ctx context.Context, cl client.Client, ns, pool, template string, replicas int32, labels map[string]string) { + p := &extv1beta1.SandboxWarmPool{} + err := cl.Get(ctx, types.NamespacedName{Namespace: ns, Name: pool}, p) + if apierrors.IsNotFound(err) { + Must(cl.Create(ctx, &extv1beta1.SandboxWarmPool{ + ObjectMeta: metav1.ObjectMeta{Name: pool, Namespace: ns, Labels: labels}, + Spec: extv1beta1.SandboxWarmPoolSpec{Replicas: &replicas, TemplateRef: extv1beta1.SandboxTemplateRef{Name: template}}, + })) + return + } + Must(err) + p.Spec.Replicas = &replicas + Must(cl.Update(ctx, p)) +} + +// ReadySandboxes counts the Sandboxes pool owns and how many are Ready. +func ReadySandboxes(ctx context.Context, cl client.Client, ns, pool string) (total, ready int) { + sl := &sandboxv1beta1.SandboxList{} + if err := cl.List(ctx, sl, client.InNamespace(ns)); err != nil { + return 0, 0 + } + for i := range sl.Items { + if !OwnedByPool(&sl.Items[i], pool) { + continue + } + total++ + for _, c := range sl.Items[i].Status.Conditions { + if c.Type == "Ready" && c.Status == metav1.ConditionTrue { + ready++ + } + } + } + return +} + +// OwnedByPool reports whether sb has an owner reference to the named warm pool. +func OwnedByPool(sb *sandboxv1beta1.Sandbox, pool string) bool { + return slices.ContainsFunc(sb.OwnerReferences, func(o metav1.OwnerReference) bool { + return o.Kind == "SandboxWarmPool" && o.Name == pool + }) +} + +// WaitReady polls until pool has target Ready members or timeoutSec elapses, +// returning the final Ready count. +func WaitReady(ctx context.Context, cl client.Client, ns, pool string, target, timeoutSec int) int { + deadline := time.Now().Add(time.Duration(timeoutSec) * time.Second) + for time.Now().Before(deadline) { + if _, ready := ReadySandboxes(ctx, cl, ns, pool); ready >= target { + return ready + } + time.Sleep(3 * time.Second) + } + _, ready := ReadySandboxes(ctx, cl, ns, pool) + return ready +} diff --git a/test/e2ebench/main.go b/test/e2ebench/main.go index d030dff..b2d5bb3 100644 --- a/test/e2ebench/main.go +++ b/test/e2ebench/main.go @@ -48,6 +48,7 @@ import ( sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) var ( @@ -73,12 +74,7 @@ const ( runVal = "g0131-phase-d" ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } func main() { flag.Parse() @@ -197,64 +193,16 @@ func ensureTemplate(ctx context.Context) { } func ensurePool(ctx context.Context, replicas int32) { - p := &extv1beta1.SandboxWarmPool{} - err := cl.Get(ctx, types.NamespacedName{Namespace: *ns, Name: poolName}, p) - if apierrors.IsNotFound(err) { - must(cl.Create(ctx, &extv1beta1.SandboxWarmPool{ - ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: *ns, Labels: map[string]string{runLabel: runVal}}, - Spec: extv1beta1.SandboxWarmPoolSpec{Replicas: &replicas, TemplateRef: extv1beta1.SandboxTemplateRef{Name: tmplName}}, - })) - return - } - must(err) - p.Spec.Replicas = &replicas - must(cl.Update(ctx, p)) + benchutil.EnsurePool(ctx, cl, *ns, poolName, tmplName, replicas, map[string]string{runLabel: runVal}) } // ourSandboxes returns the Sandbox CRs this run's pool owns and how many are Ready. func ourSandboxes(ctx context.Context) (total, ready int) { - sl := &sandboxv1beta1.SandboxList{} - if err := cl.List(ctx, sl, ctrlclient.InNamespace(*ns)); err != nil { - return 0, 0 - } - for i := range sl.Items { - if !ownedByPool(&sl.Items[i]) { - continue - } - total++ - for _, c := range sl.Items[i].Status.Conditions { - if c.Type == "Ready" && c.Status == metav1.ConditionTrue { - ready++ - } - } - } - return -} - -func ownedByPool(sb *sandboxv1beta1.Sandbox) bool { - for _, o := range sb.OwnerReferences { - if o.Kind == "SandboxWarmPool" && o.Name == poolName { - return true - } - } - return false + return benchutil.ReadySandboxes(ctx, cl, *ns, poolName) } func waitReady(ctx context.Context, target, timeoutSec int) int { - deadline := time.Now().Add(time.Duration(timeoutSec) * time.Second) - last := 0 - for time.Now().Before(deadline) { - _, ready := ourSandboxes(ctx) - if ready > last { - last = ready - } - if ready >= target { - return ready - } - time.Sleep(3 * time.Second) - } - _, ready := ourSandboxes(ctx) - return ready + return benchutil.WaitReady(ctx, cl, *ns, poolName, target, timeoutSec) } func crossCheck(ctx context.Context) (readyReplicas, sandboxCR, pods int) { diff --git a/test/l2bench/main.go b/test/l2bench/main.go index f6684ee..1b42818 100644 --- a/test/l2bench/main.go +++ b/test/l2bench/main.go @@ -28,7 +28,6 @@ import ( "net/http/httptest" "os" "path/filepath" - "sort" "strings" "sync/atomic" "time" @@ -44,6 +43,7 @@ import ( extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" "github.com/cocoonstack/sandbox-operator/pkg/scale" "github.com/cocoonstack/sandbox-operator/pkg/scale/sandboxd" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) const ( @@ -57,12 +57,7 @@ var ( orphansFlag = flag.Int("orphans", 200, "orphan bindings to inject and reconcile") ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } // fakeSandboxd is an httptest-backed sandboxd: /v1/claim hands over a fresh // sandbox fast; /v1/sandboxes/{id}/release counts VM destroys (must stay 0). @@ -134,19 +129,7 @@ func newClaim(name string) *extv1beta1.SandboxClaim { } } -func pct(xs []float64, p float64) float64 { - if len(xs) == 0 { - return 0 - } - s := append([]float64(nil), xs...) - sort.Float64s(s) - if p >= 100 { - return round4(s[len(s)-1]) - } - return round4(s[int(float64(len(s)-1)*p/100)]) -} - -func round4(f float64) float64 { return float64(int(f*10000)) / 10000 } +func pct(xs []float64, p float64) float64 { return benchutil.Pct(xs, p, benchutil.Round4) } // claimWaiter is the gateway surface the latency loop needs: Claim plus Wait to // drain async Bound records. The value scale.NewGateway returns satisfies it. diff --git a/test/l3bench/main.go b/test/l3bench/main.go index 5587a85..0bcdc25 100644 --- a/test/l3bench/main.go +++ b/test/l3bench/main.go @@ -47,6 +47,7 @@ import ( extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" "github.com/cocoonstack/sandbox-operator/pkg/scale" sandboxapiserver "github.com/cocoonstack/sandbox-operator/pkg/scale/apiserver" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) var ( @@ -57,12 +58,7 @@ var ( namespacesFlag = flag.Int("namespaces", 3, "number of namespaces to spread sandboxes across") ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } func fail(format string, args ...any) { fmt.Fprintf(os.Stderr, "FAIL: "+format+"\n", args...) diff --git a/test/poolbench/main.go b/test/poolbench/main.go index a47c9e1..0f8a66b 100644 --- a/test/poolbench/main.go +++ b/test/poolbench/main.go @@ -16,7 +16,6 @@ import ( "flag" "fmt" "os" - "sort" "strings" "sync" "time" @@ -26,7 +25,6 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/types" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" @@ -34,6 +32,7 @@ import ( sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) const image = "m.daocloud.io/docker.io/library/alpine:3.20" @@ -63,12 +62,7 @@ const ( poolName = "poolbench-pool" ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } func main() { flag.Parse() @@ -181,47 +175,11 @@ func ensureTemplate(ctx context.Context) { } func ensurePool(ctx context.Context, replicas int32) { - p := &extv1beta1.SandboxWarmPool{ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: *ns}} - err := cl.Get(ctx, types.NamespacedName{Namespace: *ns, Name: poolName}, p) - if apierrors.IsNotFound(err) { - p = &extv1beta1.SandboxWarmPool{ - ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: *ns}, - Spec: extv1beta1.SandboxWarmPoolSpec{ - Replicas: &replicas, - TemplateRef: extv1beta1.SandboxTemplateRef{Name: tmplName}, - }, - } - must(cl.Create(ctx, p)) - return - } - must(err) - p.Spec.Replicas = &replicas - must(cl.Update(ctx, p)) + benchutil.EnsurePool(ctx, cl, *ns, poolName, tmplName, replicas, nil) } func readySandboxes(ctx context.Context) (total, ready int) { - sl := &sandboxv1beta1.SandboxList{} - if err := cl.List(ctx, sl, ctrlclient.InNamespace(*ns)); err != nil { - return 0, 0 - } - for i := range sl.Items { - owned := false - for _, o := range sl.Items[i].OwnerReferences { - if o.Kind == "SandboxWarmPool" && o.Name == poolName { - owned = true - } - } - if !owned { - continue - } - total++ - for _, c := range sl.Items[i].Status.Conditions { - if c.Type == "Ready" && c.Status == metav1.ConditionTrue { - ready++ - } - } - } - return + return benchutil.ReadySandboxes(ctx, cl, *ns, poolName) } func fillPool(ctx context.Context, target int) map[string]any { @@ -263,8 +221,8 @@ func fillPool(ctx context.Context, target int) map[string]any { } return map[string]any{ "reachedReady": lastReady, - "elapsedSec": round1(elapsed), - "throughputPerSec": round2(thr), + "elapsedSec": benchutil.Round1(elapsed), + "throughputPerSec": benchutil.Round2(thr), "p50Ms": pct(c2r, 50), "p95Ms": pct(c2r, 95), "p99Ms": pct(c2r, 99), @@ -394,16 +352,4 @@ func claimLatency(ctx context.Context, n, conc int) map[string]any { } } -func pct(xs []float64, p float64) float64 { - if len(xs) == 0 { - return 0 - } - s := append([]float64(nil), xs...) - sort.Float64s(s) - if p >= 100 { - return round1(s[len(s)-1]) - } - return round1(s[int(float64(len(s)-1)*p/100)]) -} -func round1(f float64) float64 { return float64(int(f*10)) / 10 } -func round2(f float64) float64 { return float64(int(f*100)) / 100 } +func pct(xs []float64, p float64) float64 { return benchutil.Pct(xs, p, benchutil.Round1) } diff --git a/test/scalebench/main.go b/test/scalebench/main.go index d2ae29c..ba6e8b6 100644 --- a/test/scalebench/main.go +++ b/test/scalebench/main.go @@ -52,6 +52,7 @@ import ( ctrls "github.com/cocoonstack/sandbox-operator/extensions/controllers" "github.com/cocoonstack/sandbox-operator/extensions/controllers/queue" asmetrics "github.com/cocoonstack/sandbox-operator/internal/metrics" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) const ( @@ -67,12 +68,7 @@ var ( maxPasses = flag.Int("max-passes", 8, "max reconcile passes per claim before giving up") ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } func newScheme() *runtime.Scheme { s := runtime.NewScheme() @@ -126,7 +122,7 @@ func claim(idx int) *extv1beta1.SandboxClaim { // fixture returns a fake client seeded with N warm sandboxes + N claims, plus the // warm-pool queue populated with the N sandbox keys (as the event handler would). -func fixture(n int) (ctrlclient.Client, queue.SandboxQueue, []*extv1beta1.SandboxClaim, *runtime.Scheme) { +func fixture(n int) (ctrlclient.Client, *queue.SimpleSandboxQueue, []*extv1beta1.SandboxClaim, *runtime.Scheme) { scheme := newScheme() poolHash := sandboxcontrollers.NameHash(poolName) tmplHash := ctrls.SandboxTemplateRefHash(tmplName) @@ -259,19 +255,7 @@ func adoptViaList(ctx context.Context, fc ctrlclient.Client, cl *extv1beta1.Sand return false } -func pct(xs []float64, p float64) float64 { - if len(xs) == 0 { - return 0 - } - s := append([]float64(nil), xs...) - sort.Float64s(s) - if p >= 100 { - return round3(s[len(s)-1]) - } - return round3(s[int(float64(len(s)-1)*p/100)]) -} - -func round3(f float64) float64 { return float64(int(f*1000)) / 1000 } +func pct(xs []float64, p float64) float64 { return benchutil.Pct(xs, p, benchutil.Round3) } type sizeResult struct { N int `json:"n"` @@ -335,8 +319,8 @@ func main() { "substrate": "controller-runtime fake apiserver; real SandboxClaimReconciler", "sizes": sizes, "threshold_ratio": *threshold, - "fast_p50_ratio": round3(fastRatio), - "naive_p50_ratio": round3(naiveRatio), + "fast_p50_ratio": benchutil.Round3(fastRatio), + "naive_p50_ratio": benchutil.Round3(naiveRatio), "all_claims_bound": allBound, "near_constant": nearConstant, "sensitive": sensitive, diff --git a/test/scalestress/main.go b/test/scalestress/main.go index ab663ac..c9812ee 100644 --- a/test/scalestress/main.go +++ b/test/scalestress/main.go @@ -23,7 +23,6 @@ import ( "flag" "fmt" "os" - "sort" "strconv" "strings" "time" @@ -33,7 +32,6 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/clientcmd" @@ -41,6 +39,7 @@ import ( sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1" extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1" + "github.com/cocoonstack/sandbox-operator/test/benchutil" ) var ( @@ -70,12 +69,7 @@ const ( runVal = "g0131-stress" ) -func must(err error) { - if err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(2) - } -} +func must(err error) { benchutil.Must(err) } func main() { flag.Parse() @@ -157,19 +151,19 @@ func main() { "client_list_sandboxes": msStats(sbLat), "client_list_pods": msStats(podLat), // apiserver-side LIST mean over the window - "apiserver_list_sandboxes_avg_ms": round2(deltaAvgMs(prev, last, "list_sandboxes")), - "apiserver_list_pods_avg_ms": round2(deltaAvgMs(prev, last, "list_pods")), + "apiserver_list_sandboxes_avg_ms": benchutil.Round2(deltaAvgMs(prev, last, "list_sandboxes")), + "apiserver_list_pods_avg_ms": benchutil.Round2(deltaAvgMs(prev, last, "list_pods")), // the vk LIST priority level under load "vke_seats_nominal": last["vke_nominal"], - "vke_seats_inuse_peak": round2(peakInUse), - "vke_inqueue_peak": round2(peakInqueue), - "vke_wait_avg_ms": round2(deltaAvgMs(prev, last, "vke_wait")), - "vke_dispatched_delta": round0(last["vke_dispatched"] - prev["vke_dispatched"]), - "vke_rejected_timeout_delta": round0(last["vke_rej_timeout"] - prev["vke_rej_timeout"]), - "vke_rejected_cancelled_delta": round0(last["vke_rej_cancelled"] - prev["vke_rej_cancelled"]), + "vke_seats_inuse_peak": benchutil.Round2(peakInUse), + "vke_inqueue_peak": benchutil.Round2(peakInqueue), + "vke_wait_avg_ms": benchutil.Round2(deltaAvgMs(prev, last, "vke_wait")), + "vke_dispatched_delta": benchutil.Round0(last["vke_dispatched"] - prev["vke_dispatched"]), + "vke_rejected_timeout_delta": benchutil.Round0(last["vke_rej_timeout"] - prev["vke_rej_timeout"]), + "vke_rejected_cancelled_delta": benchutil.Round0(last["vke_rej_cancelled"] - prev["vke_rej_cancelled"]), "node_distribution": dist, "prod_intact": prodNow, - "window_sec": round1(curT.Sub(prevT).Seconds()), + "window_sec": benchutil.Round1(curT.Sub(prevT).Seconds()), } rounds = append(rounds, round) fmt.Printf("[N=%d] sbCR=%d pods=%d | client LIST sb p50=%.0f/p95=%.0fms pods p50=%.0f/p95=%.0fms | apiserver LIST sb=%.0fms pods=%.1fms | %s seats %.0f/%.0f peak inq=%.0f wait=%.0fms disp+%.0f REJECT_timeout+%.0f | dist=%v prod=%d\n", @@ -217,7 +211,7 @@ func main() { result["new_list_timeout_rejections"] = newRejections result["peak_vke_seats_inuse"] = maxInUse result["vke_seats_nominal"] = nominal - result["client_list_sandboxes_p95_ratio"] = round2(ratio) + result["client_list_sandboxes_p95_ratio"] = benchutil.Round2(ratio) result["wedge_observed"] = wedgeObserved // This is an exploratory probe, not an acceptance gate: "pass" just means the // run completed cleanly and did not harm production. The wedge finding is the @@ -285,60 +279,15 @@ func ensureTemplate(ctx context.Context, hosts []string) { } func setReplicas(ctx context.Context, n int32) { - p := &extv1beta1.SandboxWarmPool{} - err := cl.Get(ctx, types.NamespacedName{Namespace: *ns, Name: poolName}, p) - if apierrors.IsNotFound(err) { - must(cl.Create(ctx, &extv1beta1.SandboxWarmPool{ - ObjectMeta: metav1.ObjectMeta{Name: poolName, Namespace: *ns, Labels: map[string]string{runLabel: runVal}}, - Spec: extv1beta1.SandboxWarmPoolSpec{Replicas: &n, TemplateRef: extv1beta1.SandboxTemplateRef{Name: tmplName}}, - })) - return - } - must(err) - p.Spec.Replicas = &n - must(cl.Update(ctx, p)) + benchutil.EnsurePool(ctx, cl, *ns, poolName, tmplName, n, map[string]string{runLabel: runVal}) } func ourReady(ctx context.Context) (total, ready int) { - sl := &sandboxv1beta1.SandboxList{} - if err := cl.List(ctx, sl, ctrlclient.InNamespace(*ns)); err != nil { - return 0, 0 - } - for i := range sl.Items { - owned := false - for _, o := range sl.Items[i].OwnerReferences { - if o.Kind == "SandboxWarmPool" && o.Name == poolName { - owned = true - } - } - if !owned { - continue - } - total++ - for _, c := range sl.Items[i].Status.Conditions { - if c.Type == "Ready" && c.Status == metav1.ConditionTrue { - ready++ - } - } - } - return + return benchutil.ReadySandboxes(ctx, cl, *ns, poolName) } func waitReady(ctx context.Context, target, timeoutSec int) int { - deadline := time.Now().Add(time.Duration(timeoutSec) * time.Second) - last := 0 - for time.Now().Before(deadline) { - _, ready := ourReady(ctx) - if ready > last { - last = ready - } - if ready >= target { - return ready - } - time.Sleep(3 * time.Second) - } - _, ready := ourReady(ctx) - return ready + return benchutil.WaitReady(ctx, cl, *ns, poolName, target, timeoutSec) } func nodeDistribution(ctx context.Context) map[string]int { @@ -490,21 +439,7 @@ func msStats(s latSamples) map[string]any { } } -func pct(xs []float64, p float64) float64 { - if len(xs) == 0 { - return 0 - } - s := append([]float64(nil), xs...) - sort.Float64s(s) - if p >= 100 { - return round2(s[len(s)-1]) - } - return round2(s[int(float64(len(s)-1)*p/100)]) -} - -func round0(f float64) float64 { return float64(int(f)) } -func round1(f float64) float64 { return float64(int(f*10)) / 10 } -func round2(f float64) float64 { return float64(int(f*100)) / 100 } +func pct(xs []float64, p float64) float64 { return benchutil.Pct(xs, p, benchutil.Round2) } func splitCSV(s string) []string { var out []string