[codex] stabilize pool scaling

docs/kubernetes/index.md

@@ -378,12 +378,20 @@ spec:
     poolMin: 5
 ```
 
-Optional: add `scaleStrategy` to limit the pace of scaling:
+The pool buffer counts only unallocated pods that are Ready. Pods that are still
+starting count toward the pool's total capacity, but are not advertised as
+available buffer.
+
+Optional: add `scaleStrategy` to limit the size of each scale-up and scale-down
+batch (the default is `25%`):
 ```yaml
   scaleStrategy:
     maxUnavailable: "20%"  # or absolute number like 5
 ```
 
+The controller waits for a scale-down batch to finish terminating before it
+starts another scaling batch.
+
 Create a batch of sandboxes using the pool:
 
 ```yaml

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

@@ -36,6 +36,10 @@ spec:
       jsonPath: .status.available
       name: AVAILABLE
       type: integer
+    - description: The number of nodes updated to the latest revision.
+      jsonPath: .status.updated
+      name: UPDATED
+      type: integer
     name: v1alpha1
     schema:
       openAPIV3Schema:
@@ -169,6 +173,11 @@ spec:
                 description: Total is the total number of nodes in the pool.
                 format: int32
                 type: integer
+              updated:
+                description: Updated is the number of nodes that have been updated
+                  to the latest revision.
+                format: int32
+                type: integer
             required:
             - allocated
             - available

kubernetes/internal/controller/pool_controller.go

@@ -152,14 +152,18 @@ func (r *PoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 		log.Error(err, "Failed to list pods")
 		return reconcile.Result{}, err
 	}
+	controllerKey := controllerutils.GetControllerKey(pool)
+	podNames := make(map[string]struct{}, len(podList.Items))
 	pods := make([]*corev1.Pod, 0, len(podList.Items))
 	for i := range podList.Items {
 		pod := podList.Items[i]
-		PoolScaleExpectations.ObserveScale(controllerutils.GetControllerKey(pool), expectations.Create, pod.Name)
+		podNames[pod.Name] = struct{}{}
+		PoolScaleExpectations.ObserveScale(controllerKey, expectations.Create, pod.Name)
 		if pod.DeletionTimestamp.IsZero() {
 			pods = append(pods, &pod)
 		}
 	}
+	observeDeletedPods(controllerKey, podNames)
 
 	// List all batch sandboxes  ref to the pool
 	batchSandboxList := &sandboxv1alpha1.BatchSandboxList{}
@@ -179,11 +183,19 @@ func (r *PoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (resul
 		batchSandboxes = append(batchSandboxes, &batchSandbox)
 	}
 	log.Info("Pool reconcile", "pool", pool.Name, "pods", len(pods), "batchSandboxes", len(batchSandboxes))
-	return r.reconcilePool(ctx, pool, batchSandboxes, pods)
+	return r.reconcilePool(ctx, pool, batchSandboxes, pods, int32(len(podList.Items)))
+}
+
+func observeDeletedPods(controllerKey string, existingPodNames map[string]struct{}) {
+	for name := range PoolScaleExpectations.GetExpectations(controllerKey)[expectations.Delete] {
+		if _, exists := existingPodNames[name]; !exists {
+			PoolScaleExpectations.ObserveScale(controllerKey, expectations.Delete, name)
+		}
+	}
 }
 
 // reconcilePool contains the main reconciliation logic
-func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha1.Pool, batchSandboxes []*sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) (ctrl.Result, error) {
+func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha1.Pool, batchSandboxes []*sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod, totalPodCnt int32) (ctrl.Result, error) {
 	var result ctrl.Result
 
 	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
@@ -220,16 +232,20 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
 		args := &scaleArgs{
 			updateRevision: updateResult.UpdateRevision,
 			pods:           schedulePods,
-			totalPodCnt:    int32(len(pods)),
+			totalPodCnt:    totalPodCnt,
 			allocatedCnt:   int32(len(schedResult.LatestAllocation)),
 			idlePods:       updateResult.IdlePods,
 			toDeletePods:   toDeletePods,
 			supplyCnt:      schedResult.SupplyCnt + updateResult.SupplyUpdateRevision,
 		}
 
-		if err := r.scalePool(ctx, latestPool, args); err != nil {
+		scalePending, err := r.scalePool(ctx, latestPool, args)
+		if err != nil {
 			return err
 		}
+		if scalePending {
+			result = ctrl.Result{RequeueAfter: defaultRetryTime}
+		}
 
 		// 6. Update pool status
 		if err := r.updatePoolStatus(ctx, updateResult.UpdateRevision, latestPool, pods, schedulePods, schedResult.LatestAllocation); err != nil {
@@ -709,7 +725,7 @@ type UpdateResult struct {
 	SupplyUpdateRevision int32
 }
 
-func (r *PoolReconciler) scalePool(ctx context.Context, pool *sandboxv1alpha1.Pool, args *scaleArgs) error {
+func (r *PoolReconciler) scalePool(ctx context.Context, pool *sandboxv1alpha1.Pool, args *scaleArgs) (bool, error) {
 	log := logf.FromContext(ctx)
 	errs := make([]error, 0)
 	pods := args.pods
@@ -720,21 +736,18 @@ func (r *PoolReconciler) scalePool(ctx context.Context, pool *sandboxv1alpha1.Po
 			PoolScaleExpectations.DeleteExpectations(controllerutils.GetControllerKey(pool))
 		} else {
 			log.Info("Pool scale is not ready, requeue", "unsatisfiedDuration", unsatisfiedDuration, "dirtyPods", dirtyPods)
-			return fmt.Errorf("pool scale is not ready, %v", pool.Name)
+			return true, nil
 		}
 	}
 	schedulableCnt := int32(len(args.pods))
 	totalPodCnt := args.totalPodCnt
 	allocatedCnt := args.allocatedCnt
 	supplyCnt := args.supplyCnt
 	toDeletePods := args.toDeletePods
-	bufferCnt := schedulableCnt - allocatedCnt
+	bufferCnt := r.countReadyIdlePods(pods, args.idlePods)
 
 	// Calculate desired buffer cnt.
-	desiredBufferCnt := bufferCnt
-	if bufferCnt < pool.Spec.CapacitySpec.BufferMin || bufferCnt > pool.Spec.CapacitySpec.BufferMax {
-		desiredBufferCnt = (pool.Spec.CapacitySpec.BufferMin + pool.Spec.CapacitySpec.BufferMax) / 2
-	}
+	desiredBufferCnt := desiredBufferCount(bufferCnt, pool.Spec.CapacitySpec.BufferMin, pool.Spec.CapacitySpec.BufferMax)
 
 	// Calculate desired schedulable cnt.
 	desiredSchedulableCnt := max(allocatedCnt+supplyCnt+desiredBufferCnt, pool.Spec.CapacitySpec.PoolMin)
@@ -775,10 +788,20 @@ func (r *PoolReconciler) scalePool(ctx context.Context, pool *sandboxv1alpha1.Po
 	}
 	if scaleIn > 0 || len(toDeletePods) > 0 {
 		podsToDelete := r.pickPodsToDelete(pods, args.idlePods, args.toDeletePods, scaleIn)
+		maxDeleteCnt := r.getScaleMaxUnavailable(pool, desiredSchedulableCnt)
+		if int32(len(podsToDelete)) > maxDeleteCnt {
+			podsToDelete = podsToDelete[:maxDeleteCnt]
+		}
 		log.Info("Scaling down pool", "pool", pool.Name, "scaleIn", scaleIn, "toDeletePods", len(toDeletePods), "podsToDelete", len(podsToDelete))
 		for _, pod := range podsToDelete {
 			log.Info("Deleting pool pod", "pool", pool.Name, "pod", pod.Name)
+			controllerKey := controllerutils.GetControllerKey(pool)
+			PoolScaleExpectations.ExpectScale(controllerKey, expectations.Delete, pod.Name)
 			if err := r.Delete(ctx, pod); err != nil {
+				PoolScaleExpectations.ObserveScale(controllerKey, expectations.Delete, pod.Name)
+				if errors.IsNotFound(err) {
+					continue
+				}
 				log.Error(err, "Failed to delete pool pod", "pod", pod.Name)
 				r.Recorder.Eventf(pool, corev1.EventTypeWarning, EventReasonFailedDelete, "Failed to delete pool pod %s: %v", pod.Name, err)
 				errs = append(errs, err)
@@ -787,7 +810,7 @@ func (r *PoolReconciler) scalePool(ctx context.Context, pool *sandboxv1alpha1.Po
 			}
 		}
 	}
-	return gerrors.Join(errs...)
+	return false, gerrors.Join(errs...)
 }
 
 func (r *PoolReconciler) updatePoolStatus(ctx context.Context, updateRevision string, pool *sandboxv1alpha1.Pool, pods []*corev1.Pod, schedulePods []*corev1.Pod, podAllocation map[string]string) error {
@@ -832,38 +855,78 @@ func (r *PoolReconciler) pickPodsToDelete(pods []*corev1.Pod, idlePodNames []str
 		podMap[pod.Name] = pod
 	}
 
+	selected := make(map[string]struct{})
 	var podsToDelete []*corev1.Pod
 	for _, name := range toDeletePodNames {
 		pod, ok := podMap[name]
-		if !ok {
+		if !ok || !pod.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if _, exists := selected[name]; exists {
 			continue
 		}
 		podsToDelete = append(podsToDelete, pod)
+		selected[name] = struct{}{}
 	}
 
 	var idlePods []*corev1.Pod
 	for _, name := range idlePodNames {
 		pod, ok := podMap[name]
-		if !ok {
+		if !ok || !pod.DeletionTimestamp.IsZero() {
 			continue
 		}
 		idlePods = append(idlePods, pod)
 	}
 	sort.Slice(idlePods, func(i, j int) bool {
-		return idlePods[i].CreationTimestamp.Before(&idlePods[j].CreationTimestamp)
+		iReady := utils.IsPodReady(idlePods[i])
+		jReady := utils.IsPodReady(idlePods[j])
+		if iReady != jReady {
+			return !iReady
+		}
+		if idlePods[i].CreationTimestamp.Equal(&idlePods[j].CreationTimestamp) {
+			return idlePods[i].Name < idlePods[j].Name
+		}
+		return idlePods[i].CreationTimestamp.After(idlePods[j].CreationTimestamp.Time)
 	})
 	for _, pod := range idlePods {
 		if scaleIn <= 0 {
 			break
 		}
-		if pod.DeletionTimestamp == nil {
-			podsToDelete = append(podsToDelete, pod)
+		if _, exists := selected[pod.Name]; exists {
+			continue
 		}
+		podsToDelete = append(podsToDelete, pod)
+		selected[pod.Name] = struct{}{}
 		scaleIn -= 1
 	}
 	return podsToDelete
 }
 
+func desiredBufferCount(readyBufferCnt, bufferMin, bufferMax int32) int32 {
+	if readyBufferCnt < bufferMin {
+		return bufferMin
+	}
+	if readyBufferCnt > bufferMax {
+		return bufferMax
+	}
+	return readyBufferCnt
+}
+
+func (r *PoolReconciler) countReadyIdlePods(pods []*corev1.Pod, idlePodNames []string) int32 {
+	idleNames := make(map[string]struct{}, len(idlePodNames))
+	for _, name := range idlePodNames {
+		idleNames[name] = struct{}{}
+	}
+
+	var count int32
+	for _, pod := range pods {
+		if _, idle := idleNames[pod.Name]; idle && utils.IsPodReady(pod) {
+			count++
+		}
+	}
+	return count
+}
+
 // getScaleMaxUnavailable returns the resolved maxUnavailable value.
 // If not specified, defaults to 25% of desiredTotal.
 // Minimum return value is 1 to ensure scaling progress.