fix(server): surface Pool capacity exhaustion

docs/kubernetes/index.md

@@ -399,6 +399,17 @@ spec:
   poolRef: example-pool
 ```
 
+::: info Pool capacity back-pressure
+When a lifecycle request cannot obtain a slot because the selected Pool is at
+`poolMax`, the controller records `PoolAllocationPending=True` with reason
+`PoolCapacityExhausted` on the BatchSandbox. The Server waits up to
+`kubernetes.pool_acquisition_timeout_seconds` (30 seconds by default), without
+extending the overall sandbox creation timeout. If capacity remains unavailable,
+the request returns HTTP `429`, error code
+`KUBERNETES::POOL_CAPACITY_EXHAUSTED`, and a `Retry-After` header. A slot released
+during the acquisition window can still satisfy the request.
+:::
+
 ::: warning Per-request network policies
 Pool pods are created before allocation. The lifecycle API therefore rejects `networkPolicy` together with `extensions.poolRef`; it cannot inject an egress sidecar into an existing pool pod. Configure required network controls in the Pool pod template before pods are created, or use a non-pooled sandbox for per-request policies.
 :::

kubernetes/apis/sandbox/v1alpha1/batchsandbox_types.go

@@ -41,7 +41,7 @@ const (
 )
 
 // BatchSandboxConditionType represents the type of BatchSandbox condition.
-// +kubebuilder:validation:Enum=Ready;Progressing;Paused;PauseFailed;ResumeFailed;PodFailed
+// +kubebuilder:validation:Enum=Ready;Progressing;Paused;PauseFailed;ResumeFailed;PodFailed;PoolAllocationPending
 type BatchSandboxConditionType string
 
 const (
@@ -57,6 +57,8 @@ const (
 	BatchSandboxConditionResumeFailed BatchSandboxConditionType = "ResumeFailed"
 	// BatchSandboxConditionPodFailed is set when the sandbox pod enters a failed state.
 	BatchSandboxConditionPodFailed BatchSandboxConditionType = "PodFailed"
+	// BatchSandboxConditionPoolAllocationPending is set while Pool capacity prevents allocation.
+	BatchSandboxConditionPoolAllocationPending BatchSandboxConditionType = "PoolAllocationPending"
 )
 
 // BatchSandboxCondition represents a condition of a BatchSandbox
@@ -182,7 +184,7 @@ type BatchSandboxStatus struct {
 	// +optional
 	PauseObservedGeneration int64 `json:"pauseObservedGeneration,omitempty"`
 
-	// Conditions records operation failure context
+	// Conditions records lifecycle and operational state details.
 	// +optional
 	// +listType=map
 	// +listMapKey=type

kubernetes/charts/opensandbox-controller/templates/crds/batchsandboxes.yaml

@@ -156,7 +156,7 @@ spec:
                 format: int32
                 type: integer
               conditions:
-                description: Conditions records operation failure context
+                description: Conditions records lifecycle and operational state details.
                 items:
                   description: BatchSandboxCondition represents a condition of a BatchSandbox
                   properties:
@@ -186,6 +186,7 @@ spec:
                       - PauseFailed
                       - ResumeFailed
                       - PodFailed
+                      - PoolAllocationPending
                       type: string
                   required:
                   - status

kubernetes/config/crd/bases/sandbox.opensandbox.io_batchsandboxes.yaml

@@ -147,7 +147,7 @@ spec:
                 format: int32
                 type: integer
               conditions:
-                description: Conditions records operation failure context
+                description: Conditions records lifecycle and operational state details.
                 items:
                   description: BatchSandboxCondition represents a condition of a BatchSandbox
                   properties:
@@ -177,6 +177,7 @@ spec:
                       - PauseFailed
                       - ResumeFailed
                       - PodFailed
+                      - PoolAllocationPending
                       type: string
                   required:
                   - status

kubernetes/internal/controller/batchsandbox_controller.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -57,7 +58,13 @@ var (
 	DurationStore                 = requeueduration.DurationStore{}
 )
 
-const batchSandboxFirstPodIndex = 0
+const (
+	batchSandboxFirstPodIndex = 0
+	poolAllocationRetryTime   = 5 * time.Second
+	poolAutoAssignRef         = "*"
+)
+
+const poolCapacityExhaustedReason = poolassign.FailureCodeCapacityExhausted
 
 type taskScheduleResult struct {
 	Running, Failed, Succeed, Unknown, Pending int32
@@ -137,9 +144,24 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	if profileName := poolStrategy.AssignProfile(); profileName != "" {
 		updated, err := r.assignPool(ctx, batchSbx, profileName)
 		if err != nil {
+			var noEligiblePoolErr *poolassign.NoEligiblePoolError
+			if gerrors.As(err, &noEligiblePoolErr) && noEligiblePoolErr.CapacityExhausted() {
+				if statusErr := r.setPoolAllocationPending(
+					ctx,
+					batchSbx,
+					true,
+					"All otherwise eligible Pools are at capacity",
+				); statusErr != nil {
+					return ctrl.Result{}, fmt.Errorf("failed to publish pool capacity status: %w", statusErr)
+				}
+				return ctrl.Result{RequeueAfter: poolAllocationRetryTime}, nil
+			}
 			return ctrl.Result{}, fmt.Errorf("failed to auto-assign pool: %w", err)
 		}
 		if updated {
+			if err := r.setPoolAllocationPending(ctx, batchSbx, false, ""); err != nil {
+				return ctrl.Result{}, fmt.Errorf("failed to clear pool capacity status: %w", err)
+			}
 			return ctrl.Result{}, nil
 		}
 	}
@@ -196,6 +218,21 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	runtimeView := buildRuntimeView(batchSbx, pods)
+	poolAllocationPending, err := r.applyFixedPoolCapacityCondition(ctx, batchSbx, runtimeView.status)
+	if err != nil {
+		aggErrors = append(aggErrors, err)
+	}
+	poolAllocationActive := runtimeView.status.Phase == "" ||
+		runtimeView.status.Phase == sandboxv1alpha1.BatchSandboxPhasePending
+	if poolAllocationActive &&
+		poolStrategy.IsPooledMode() &&
+		batchSbx.Spec.Replicas != nil &&
+		runtimeView.status.Allocated < *batchSbx.Spec.Replicas {
+		DurationStore.Push(req.String(), poolAllocationRetryTime)
+	}
+	if poolAllocationPending {
+		log.Info("Sandbox is waiting for Pool capacity", "pool", batchSbx.Spec.PoolRef)
+	}
 	// Ensure PauseObservedGeneration is up-to-date so the status patch ACKs the
 	// current generation without requiring a dedicated API call.
 	// Skip during Resuming: a newer generation may carry a queued pause request
@@ -235,6 +272,105 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	return reconcile.Result{RequeueAfter: requeueAfter}, gerrors.Join(aggErrors...)
 }
 
+func (r *BatchSandboxReconciler) setPoolAllocationPending(
+	ctx context.Context,
+	batchSbx *sandboxv1alpha1.BatchSandbox,
+	pending bool,
+	message string,
+) error {
+	newStatus := batchSbx.Status.DeepCopy()
+	conditionStatus := sandboxv1alpha1.ConditionFalse
+	reason := ""
+	if pending {
+		conditionStatus = sandboxv1alpha1.ConditionTrue
+		reason = poolCapacityExhaustedReason
+	}
+	setConditionInStatus(
+		newStatus,
+		sandboxv1alpha1.BatchSandboxConditionPoolAllocationPending,
+		conditionStatus,
+		reason,
+		message,
+	)
+	if equality.Semantic.DeepEqual(*newStatus, batchSbx.Status) {
+		return nil
+	}
+	return r.updateStatusConditions(ctx, batchSbx, newStatus.Conditions)
+}
+
+func (r *BatchSandboxReconciler) applyFixedPoolCapacityCondition(
+	ctx context.Context,
+	batchSbx *sandboxv1alpha1.BatchSandbox,
+	status *sandboxv1alpha1.BatchSandboxStatus,
+) (bool, error) {
+	if status.Phase != "" && status.Phase != sandboxv1alpha1.BatchSandboxPhasePending {
+		setConditionInStatus(
+			status,
+			sandboxv1alpha1.BatchSandboxConditionPoolAllocationPending,
+			sandboxv1alpha1.ConditionFalse,
+			"",
+			"",
+		)
+		return false, nil
+	}
+	if batchSbx.Spec.PoolRef == "" || batchSbx.Spec.PoolRef == poolAutoAssignRef {
+		setConditionInStatus(
+			status,
+			sandboxv1alpha1.BatchSandboxConditionPoolAllocationPending,
+			sandboxv1alpha1.ConditionFalse,
+			"",
+			"",
+		)
+		return false, nil
+	}
+
+	desired := int32(1)
+	if batchSbx.Spec.Replicas != nil {
+		desired = *batchSbx.Spec.Replicas
+	}
+	remaining := desired - status.Allocated
+	if remaining <= 0 {
+		setConditionInStatus(
+			status,
+			sandboxv1alpha1.BatchSandboxConditionPoolAllocationPending,
+			sandboxv1alpha1.ConditionFalse,
+			"",
+			"",
+		)
+		return false, nil
+	}
+
+	pool := &sandboxv1alpha1.Pool{}
+	if err := r.Get(ctx, client.ObjectKey{Namespace: batchSbx.Namespace, Name: batchSbx.Spec.PoolRef}, pool); err != nil {
+		return false, fmt.Errorf("failed to inspect pool %s/%s capacity: %w", batchSbx.Namespace, batchSbx.Spec.PoolRef, err)
+	}
+
+	available := max(pool.Spec.CapacitySpec.PoolMax-pool.Status.Allocated, 0)
+	exhausted := available < remaining
+	conditionStatus := sandboxv1alpha1.ConditionFalse
+	reason := ""
+	message := ""
+	if exhausted {
+		conditionStatus = sandboxv1alpha1.ConditionTrue
+		reason = poolCapacityExhaustedReason
+		message = fmt.Sprintf(
+			"Pool %s has insufficient capacity for %d remaining replica(s): poolMax=%d, allocated=%d",
+			pool.Name,
+			remaining,
+			pool.Spec.CapacitySpec.PoolMax,
+			pool.Status.Allocated,
+		)
+	}
+	setConditionInStatus(
+		status,
+		sandboxv1alpha1.BatchSandboxConditionPoolAllocationPending,
+		conditionStatus,
+		reason,
+		message,
+	)
+	return exhausted, nil
+}
+
 func calPodIndex(poolStrategy strategy.PoolStrategy, batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) (map[string]int, error) {
 	podIndex := map[string]int{}
 	if poolStrategy.IsPooledMode() {