review: whole-repo round — owning-node index, cache-fed warm pool, scoped informers, indexed queue

Makefile

@@ -138,3 +138,7 @@ cloc: ## Count lines of code excluding tests (requires cloc).
 .PHONY: help
 help: ## Show available targets.
 	@awk 'BEGIN {FS = ":.*## "; printf "Usage: make <target>\n\nTargets:\n"} /^[a-zA-Z0-9_-]+:.*## / {printf "  %-14s %s\n", $$1, $$2}' $(MAKEFILE_LIST)
+
+.PHONY: api-docs
+api-docs: ## Regenerate docs/api.md from the API types
+	GOWORK=off go run github.com/elastic/crd-ref-docs@v0.2.0 --config=hack/crd-ref-docs.yaml --source-path=. --renderer=markdown --output-path=docs/api.md --max-depth=12

api/v1alpha1/sandbox_conversion.go

@@ -54,7 +54,7 @@ func (s *Sandbox) ConvertTo(dstRaw conversion.Hub) error {
 	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)
+		return fmt.Errorf("marshal v1alpha1 sandbox state: %w", err)
 	}
 	dst.Annotations[v1alpha1SandboxStateAnnotation] = string(stateJSON)
 
@@ -84,7 +84,7 @@ func (s *Sandbox) ConvertFrom(srcRaw conversion.Hub) error {
 
 		var original v1alpha1State
 		if err := json.Unmarshal([]byte(stateJSON), &original); err != nil {
-			return fmt.Errorf("failed to unmarshal v1alpha1 Sandbox state: %w", err)
+			return fmt.Errorf("unmarshal v1alpha1 sandbox state: %w", err)
 		}
 
 		// Restore replicas field from original if OperatingMode matches original intent
@@ -119,7 +119,6 @@ func ConvertSpecTo(src *SandboxSpec, dst *v1beta1.SandboxSpec) {
 
 	ConvertLifecycleTo(&src.Lifecycle, &dst.Lifecycle)
 
-	// Replicas -> OperatingMode
 	if src.Replicas != nil && *src.Replicas == 0 {
 		dst.OperatingMode = v1beta1.SandboxOperatingModeSuspended
 	} else {
@@ -143,7 +142,6 @@ func ConvertSpecFrom(src *v1beta1.SandboxSpec, dst *SandboxSpec) {
 
 	ConvertLifecycleFrom(&src.Lifecycle, &dst.Lifecycle)
 
-	// OperatingMode -> Replicas
 	if src.OperatingMode == v1beta1.SandboxOperatingModeSuspended {
 		dst.Replicas = new(int32(0))
 	} else {

api/v1alpha1/sandbox_conversion_bench_test.go

@@ -10,10 +10,6 @@ import (
 	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()

api/v1alpha1/sandbox_conversion_test.go

@@ -64,7 +64,6 @@ func TestSandboxConversion(t *testing.T) {
 			policy := ShutdownPolicyDelete
 			bTrue := true
 
-			// Create src v1alpha1 Sandbox
 			src := &Sandbox{
 				Name:      "my-sandbox",
 				Namespace: "default",
@@ -118,13 +117,11 @@ func TestSandboxConversion(t *testing.T) {
 				},
 			}
 
-			// Convert to Hub (v1beta1)
 			dst := &v1beta1.Sandbox{}
 			if err := src.ConvertTo(dst); err != nil {
 				t.Fatalf("failed to convert to v1beta1: %v", err)
 			}
 
-			// Verify src annotations and labels were not mutated during ConvertTo
 			if val, ok := src.Annotations[v1alpha1SandboxStateAnnotation]; !ok || val != "some-old-state" {
 				t.Errorf("src.Annotations was mutated during ConvertTo! expected 'some-old-state', got %q", val)
 			}
@@ -135,7 +132,6 @@ func TestSandboxConversion(t *testing.T) {
 				t.Errorf("expected 1 label in src, got %d", len(src.Labels))
 			}
 
-			// Verify the marshaled state in dst does not contain the state annotation itself (no nesting)
 			marshaledState := dst.Annotations[v1alpha1SandboxStateAnnotation]
 			var stateObj Sandbox
 			if err := json.Unmarshal([]byte(marshaledState), &stateObj); err != nil {
@@ -145,7 +141,6 @@ func TestSandboxConversion(t *testing.T) {
 				t.Errorf("dst.Annotations state nestedly contains the state annotation! causing exponential growth")
 			}
 
-			// Verify fields in v1beta1
 			if dst.Spec.OperatingMode != tc.expectedMode {
 				t.Errorf("expected OperatingMode %q, got %q", tc.expectedMode, dst.Spec.OperatingMode)
 			}
@@ -159,18 +154,15 @@ func TestSandboxConversion(t *testing.T) {
 				t.Errorf("expected ShutdownPolicy %q, got %v", ShutdownPolicyDelete, dst.Spec.ShutdownPolicy)
 			}
 
-			// Convert back to Spoke (v1alpha1)
 			roundTrip := &Sandbox{}
 			if err := roundTrip.ConvertFrom(dst); err != nil {
 				t.Fatalf("failed to convert back to v1alpha1: %v", err)
 			}
 
-			// Verify state annotation was stripped during ConvertFrom
 			if _, ok := roundTrip.Annotations[v1alpha1SandboxStateAnnotation]; ok {
 				t.Errorf("roundTrip.Annotations still contains the state annotation after ConvertFrom!")
 			}
 
-			// Verify round-trip preserves fields losslessly
 			if tc.replicas == nil {
 				if roundTrip.Spec.Replicas != nil {
 					t.Errorf("roundtrip Replicas mismatch: expected nil, got %v", *roundTrip.Spec.Replicas)
@@ -200,9 +192,6 @@ 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{
@@ -232,7 +221,6 @@ func TestConvertFromLegacyFullObjectState(t *testing.T) {
 }
 
 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 {
 		name             string
 		mode             v1beta1.SandboxOperatingMode

api/v1alpha1/sandbox_types.go

@@ -53,8 +53,6 @@ const (
 	SandboxPodNameAnnotation = "agents.x-k8s.io/pod-name"
 	// SandboxTemplateRefAnnotation is the annotation used to track the sandbox template ref.
 	SandboxTemplateRefAnnotation = "agents.x-k8s.io/sandbox-template-ref"
-	// SandboxPodTemplateHashLabel is the label used to track the pod template hash.
-	SandboxPodTemplateHashLabel = "agents.x-k8s.io/sandbox-pod-template-hash"
 	// SandboxPropagatedLabelsAnnotation is the annotation used to track the labels explicitly propagated from sandbox spec to pod.
 	SandboxPropagatedLabelsAnnotation = "agents.x-k8s.io/propagated-labels"
 	// SandboxPropagatedAnnotationsAnnotation is the annotation used to track the annotations explicitly propagated from sandbox spec to pod.

cmd/sandbox-apiserver/main.go

@@ -183,7 +183,7 @@ func (o *options) serverConfig() (*genericapiserver.Config, error) {
 // apiserver's inventory cache: the warm-pool driver and the e2b REST surface.
 func (o *options) startSidecars(ctx context.Context, restCfg *restclient.Config, token string, store scale.SandboxStore, invSource scale.InventorySource) error {
 	if o.WarmPoolDriver {
-		if err := startWarmPoolDriver(ctx, restCfg, token, o.WarmPoolInterval); err != nil {
+		if err := startWarmPoolDriver(ctx, restCfg, token, o.WarmPoolInterval, invSource); err != nil {
 			return err
 		}
 	}
@@ -294,8 +294,10 @@ func startInventoryCache(ctx context.Context, restCfg *restclient.Config) (cache
 // a poll tick (the only latency that ever mattered — the node side fills a pool
 // in under a second). Leader election makes exactly one of the apiserver replicas
 // drive the pools. The manager's own metrics/health servers are disabled; the
-// aggregated apiserver owns the serving port.
-func startWarmPoolDriver(ctx context.Context, restCfg *restclient.Config, token string, interval time.Duration) error {
+// aggregated apiserver owns the serving port. inv is the process-wide cache-fed
+// inventory source; the manager's own client would read NodeInventory
+// unstructured and so bypass its cache on every node read.
+func startWarmPoolDriver(ctx context.Context, restCfg *restclient.Config, token string, interval time.Duration, inv scale.InventorySource) error {
 	scheme := runtime.NewScheme()
 	if err := extv1beta1.AddToScheme(scheme); err != nil {
 		return fmt.Errorf("register extensions scheme: %w", err)
@@ -311,7 +313,7 @@ func startWarmPoolDriver(ctx context.Context, restCfg *restclient.Config, token
 	if err != nil {
 		return fmt.Errorf("build warm-pool manager: %w", err)
 	}
-	driver := warmpool.New(nil, nil, token, warmpool.NewSandboxdFactory(), warmpool.Options{
+	driver := warmpool.New(nil, inv, token, warmpool.NewSandboxdFactory(), warmpool.Options{
 		Interval: interval,
 		Log:      ctrl.Log.WithName("warmpool"),
 	})

cmd/sandbox-loadgen/main.go

@@ -37,16 +37,20 @@ import (
 
 	sandboxv1beta1 "github.com/cocoonstack/sandbox-operator/api/v1beta1"
 	extv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1"
+	asmetrics "github.com/cocoonstack/sandbox-operator/internal/metrics"
+	"github.com/cocoonstack/sandbox-operator/pkg/podruntime"
 )
 
 const (
-	runtimeAnnotation = "sandbox.cocoonstack.io/runtime"
+	runtimeAnnotation = podruntime.RuntimeAnnotation
 	templateName      = "loadgen-tpl"
-	warmPoolName      = "loadgen-pool"
+	// releaseTimeout bounds the detached release so shutdown cannot hang on it.
+	releaseTimeout = 10 * time.Second
+	warmPoolName   = "loadgen-pool"
 	// observedAtAnnotation is read by the operator to compute the claim-startup
 	// latency (time.Since). Stamping it at create-time makes the operator record
 	// agent_sandbox_claim_startup_latency_ms for our claims.
-	observedAtAnnotation = "agents.x-k8s.io/controller-first-observed-at"
+	observedAtAnnotation = asmetrics.ObservabilityAnnotation
 )
 
 var (
@@ -293,6 +297,19 @@ func (l *loadgen) claimOnce(ctx context.Context, name string) {
 		return
 	}
 	claimsTotal.Inc()
+	// Deferred and detached from ctx: an early return on shutdown would otherwise
+	// strand the claim, and with it the warm sandbox, until its own expiry.
+	defer func() {
+		rel := &unstructured.Unstructured{}
+		rel.SetGroupVersionKind(gvk("SandboxClaim"))
+		rel.SetNamespace(l.o.namespace)
+		rel.SetName(name)
+		relCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), releaseTimeout)
+		defer cancel()
+		if err := l.cl.Delete(relCtx, rel); err != nil && !apierrors.IsNotFound(err) {
+			claimFailed.WithLabelValues("delete").Inc()
+		}
+	}()
 
 	poll := time.NewTicker(l.o.poll)
 	defer poll.Stop()
@@ -320,14 +337,6 @@ func (l *loadgen) claimOnce(ctx context.Context, name string) {
 			break
 		}
 	}
-	// release
-	rel := &unstructured.Unstructured{}
-	rel.SetGroupVersionKind(gvk("SandboxClaim"))
-	rel.SetNamespace(l.o.namespace)
-	rel.SetName(name)
-	if err := l.cl.Delete(ctx, rel); err != nil && !apierrors.IsNotFound(err) {
-		claimFailed.WithLabelValues("delete").Inc()
-	}
 }
 
 func (l *loadgen) poolPollLoop(ctx context.Context) {

cmd/sandbox-operator/main.go

@@ -27,16 +27,14 @@ import (
 	"strings"
 	"time"
 
-	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
-	// to ensure that exec-entrypoint and run can make use of them.
-	_ "k8s.io/client-go/plugin/pkg/client/auth"
-
 	"github.com/felixge/fgprof"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	_ "k8s.io/client-go/plugin/pkg/client/auth" // every kubeconfig auth plugin an operator may be handed
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -48,8 +46,8 @@ import (
 	extensionsv1alpha1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1alpha1"
 	extensionsv1beta1 "github.com/cocoonstack/sandbox-operator/extensions/api/v1beta1"
 	extensionscontrollers "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/internal/queue"
 	"github.com/cocoonstack/sandbox-operator/internal/version"
 	"github.com/cocoonstack/sandbox-operator/pkg/podruntime"
 	//+kubebuilder:scaffold:imports
@@ -188,8 +186,7 @@ func (o *options) run() error {
 	}
 	defer cleanup()
 
-	// Importing net/http/pprof registers handlers on the global DefaultServeMux.
-	// Reset it so no server using the default mux exposes pprof by accident.
+	// net/http/pprof registers on DefaultServeMux at import time, so reset it.
 	http.DefaultServeMux = http.NewServeMux()
 
 	scheme := controllers.Scheme
@@ -207,8 +204,13 @@ func (o *options) run() error {
 		return fmt.Errorf("webhook certificate setup: %w", err)
 	}
 
+	cacheByObject, err := controllers.CacheByObject()
+	if err != nil {
+		return err
+	}
 	mgr, err := ctrl.NewManager(restConfig, ctrl.Options{
 		Scheme:                  scheme,
+		Cache:                   cache.Options{ByObject: cacheByObject},
 		Metrics:                 metricsserver.Options{BindAddress: o.metricsAddr, ExtraHandlers: o.pprofHandlers()},
 		HealthProbeBindAddress:  o.probeAddr,
 		LeaderElection:          o.enableLeaderElection,