fix(pool): skip recycling pods in allocate actions

Author: fengcone

kubernetes/internal/controller/allocator.go

@@ -227,10 +227,6 @@ func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec
 		if spec.RecyclingPods.Has(pod.Name) {
 			continue
 		}
-		// Exclude pods that are restarting (have recycle-meta annotation)
-		if isRecycling(pod) {
-			continue
-		}
 		if pod.Status.Phase != corev1.PodRunning {
 			continue
 		}
@@ -240,11 +236,11 @@ func (allocator *defaultAllocator) Schedule(ctx context.Context, spec *AllocSpec
 	for podName, sandboxName := range status.PodAllocation {
 		sandboxToPods[sandboxName] = append(sandboxToPods[sandboxName], podName)
 	}
-	sandboxAlloc, dirtySandboxes, poolAllocate, err := allocator.allocate(ctx, status, sandboxToPods, availablePods, spec.Sandboxes, spec.Pods)
+	sandboxAlloc, dirtySandboxes, poolAllocate, err := allocator.allocate(ctx, status, sandboxToPods, availablePods, spec.Sandboxes)
 	if err != nil {
 		log.Error(err, "allocate failed")
 	}
-	poolDeallocate, err := allocator.deallocate(ctx, status, sandboxToPods, spec.Sandboxes)
+	poolDeallocate, err := allocator.deallocate(ctx, status, sandboxToPods, spec.Sandboxes, spec.RecyclingPods)
 	if err != nil {
 		log.Error(err, "deallocate failed")
 	}
@@ -284,7 +280,7 @@ func (allocator *defaultAllocator) initAllocation(ctx context.Context, spec *All
 	return status, nil
 }
 
-func (allocator *defaultAllocator) allocate(ctx context.Context, status *AllocStatus, sandboxToPods map[string][]string, availablePods []string, sandboxes []*sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) (map[string][]string, []string, bool, error) {
+func (allocator *defaultAllocator) allocate(ctx context.Context, status *AllocStatus, sandboxToPods map[string][]string, availablePods []string, sandboxes []*sandboxv1alpha1.BatchSandbox) (map[string][]string, []string, bool, error) {
 	errs := make([]error, 0)
 	sandboxAlloc := make(map[string][]string)
 	dirtySandboxes := make([]string, 0)
@@ -366,7 +362,7 @@ func (allocator *defaultAllocator) doAllocate(ctx context.Context, status *Alloc
 	return sandboxAlloc, remainAvailablePods, sandboxDirty, poolAllocate, nil
 }
 
-func (allocator *defaultAllocator) deallocate(ctx context.Context, status *AllocStatus, sandboxToPods map[string][]string, sandboxes []*sandboxv1alpha1.BatchSandbox) (bool, error) {
+func (allocator *defaultAllocator) deallocate(ctx context.Context, status *AllocStatus, sandboxToPods map[string][]string, sandboxes []*sandboxv1alpha1.BatchSandbox, recycling sets.Set[string]) (bool, error) {
 	log := logf.FromContext(ctx)
 	poolDeallocate := false
 	errs := make([]error, 0)
@@ -393,6 +389,9 @@ func (allocator *defaultAllocator) deallocate(ctx context.Context, status *Alloc
 		pods := sandboxToPods[name]
 		log.Info("GC deleted sandbox allocation", "sandbox", name, "podCount", len(pods))
 		for _, pod := range pods {
+			if recycling.Has(pod) {
+				continue
+			}
 			delete(status.PodAllocation, pod)
 			poolDeallocate = true
 		}

kubernetes/internal/controller/allocator_test.go

@@ -22,6 +22,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1"
 	"github.com/golang/mock/gomock"
@@ -324,6 +325,7 @@ func TestAllocatorSchedule(t *testing.T) {
 						Name: "pool1",
 					},
 				},
+				RecyclingPods: sets.New("pod-deallocated"),
 				Sandboxes: []*sandboxv1alpha1.BatchSandbox{
 					{
 						ObjectMeta: metav1.ObjectMeta{
@@ -610,9 +612,10 @@ func TestScheduleExcludesRestartingPods(t *testing.T) {
 		},
 	}
 	spec := &AllocSpec{
-		Pods:      pods,
-		Sandboxes: sandboxes,
-		Pool:      &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool1"}},
+		Pods:          pods,
+		Sandboxes:     sandboxes,
+		Pool:          &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool1"}},
+		RecyclingPods: sets.New("pod-restarting"),
 	}
 
 	store.EXPECT().GetAllocation(gomock.Any(), gomock.Any()).Return(&PoolAllocation{PodAllocation: map[string]string{}}, nil).Times(1)

kubernetes/internal/controller/batchsandbox_controller.go

@@ -622,8 +622,14 @@ func (r *BatchSandboxReconciler) addDeallocatedFromLabel(ctx context.Context, bs
 	if err != nil {
 		return err
 	}
+	released, err := parseSandboxReleased(bsx)
+	if err != nil {
+		return err
+	}
+	pods := sets.NewString(released.Pods...)
+	pods.Insert(alloc.Pods...)
 
-	for _, podName := range alloc.Pods {
+	for podName := range pods {
 		pod := &corev1.Pod{}
 		err = r.Get(ctx, types.NamespacedName{Namespace: bsx.Namespace, Name: podName}, pod)
 		if errors.IsNotFound(err) {

kubernetes/internal/controller/pool_controller.go

@@ -158,7 +158,7 @@ func (r *PoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 	return r.reconcilePool(ctx, pool, batchSandboxes, pods)
 }
 
-func (r *PoolReconciler) collectRecyclingPods(batchSandboxes []*sandboxv1alpha1.BatchSandbox) (sets.Set[string], error) {
+func (r *PoolReconciler) collectRecyclingPods(batchSandboxes []*sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) (sets.Set[string], error) {
 	recycling := make(sets.Set[string])
 	for _, batchSandbox := range batchSandboxes {
 		if batchSandbox.DeletionTimestamp != nil {
@@ -178,6 +178,11 @@ func (r *PoolReconciler) collectRecyclingPods(batchSandboxes []*sandboxv1alpha1.
 			}
 		}
 	}
+	for _, pod := range pods {
+		if isRecycling(pod) {
+			recycling.Insert(pod.Name)
+		}
+	}
 	return recycling, nil
 }
 
@@ -194,7 +199,7 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
 		}
 
 		// 2. First, handle Pod Recycle to ensure pods are ready for scheduling
-		recyclingPods, err := r.collectRecyclingPods(batchSandboxes)
+		recyclingPods, err := r.collectRecyclingPods(batchSandboxes, pods)
 		if err != nil {
 			log.Error(err, "Failed to collect recycling pods")
 			return err
@@ -250,7 +255,7 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
 			pool:           latestPool,
 			pods:           pods,
 			allocatedCnt:   int32(len(scheRes.podAllocation)),
-			recycling:      int32(len(recyclingPods)),
+			recycling:      recyclingPods,
 			idlePods:       latestIdlePods,
 			redundantPods:  deleteOld,
 			supplyCnt:      scheRes.supplySandbox + supplyNew,
@@ -310,6 +315,9 @@ func (r *PoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			if oldObj.Spec.Replicas != newObj.Spec.Replicas {
 				return true
 			}
+			if oldObj.DeletionTimestamp == nil && newObj.DeletionTimestamp != nil {
+				return true
+			}
 			return false
 		},
 		DeleteFunc: func(e event.DeleteEvent) bool {
@@ -425,8 +433,8 @@ type scaleArgs struct {
 	pool           *sandboxv1alpha1.Pool
 	pods           []*corev1.Pod
 	allocatedCnt   int32
-	recycling      int32 // pods that are restarting and not available
-	supplyCnt      int32 // to create
+	recycling      sets.Set[string] // pods that are restarting and not available
+	supplyCnt      int32            // to create
 	idlePods       []string
 	redundantPods  []string
 }
@@ -446,7 +454,7 @@ func (r *PoolReconciler) scalePool(ctx context.Context, args *scaleArgs) error {
 	supplyCnt := args.supplyCnt
 	redundantPods := args.redundantPods
 	// Buffer count excludes allocated and restarting pods
-	bufferCnt := totalCnt - allocatedCnt - recycling
+	bufferCnt := totalCnt - allocatedCnt - int32(len(recycling))
 
 	// Calculate desired buffer cnt.
 	desiredBufferCnt := bufferCnt
@@ -482,7 +490,7 @@ func (r *PoolReconciler) scalePool(ctx context.Context, args *scaleArgs) error {
 		if desiredTotalCnt < totalCnt {
 			scaleIn = totalCnt - desiredTotalCnt
 		}
-		podsToDelete := r.pickPodsToDelete(pods, args.idlePods, args.redundantPods, scaleIn)
+		podsToDelete := r.pickPodsToDelete(pods, args.idlePods, args.redundantPods, scaleIn, args.recycling)
 		log.Info("Scaling down pool", "pool", pool.Name, "scaleIn", scaleIn, "redundantPods", len(redundantPods), "podsToDelete", len(podsToDelete))
 		for _, pod := range podsToDelete {
 			log.Info("Deleting pool pod", "pool", pool.Name, "pod", pod.Name)
@@ -532,7 +540,7 @@ func (r *PoolReconciler) updatePoolStatus(ctx context.Context, latestRevision st
 	return nil
 }
 
-func (r *PoolReconciler) pickPodsToDelete(pods []*corev1.Pod, idlePodNames []string, redundantPodNames []string, scaleIn int32) []*corev1.Pod {
+func (r *PoolReconciler) pickPodsToDelete(pods []*corev1.Pod, idlePodNames []string, redundantPodNames []string, scaleIn int32, recycling sets.Set[string]) []*corev1.Pod {
 	var idlePods []*corev1.Pod
 	podMap := make(map[string]*corev1.Pod)
 	for _, pod := range pods {
@@ -555,12 +563,18 @@ func (r *PoolReconciler) pickPodsToDelete(pods []*corev1.Pod, idlePodNames []str
 		if !ok {
 			continue
 		}
+		if recycling.Has(pod.Name) {
+			continue
+		}
 		podsToDelete = append(podsToDelete, pod)
 	}
 	for _, pod := range idlePods { // delete pod from pool scale
 		if scaleIn <= 0 {
 			break
 		}
+		if recycling.Has(pod.Name) {
+			continue
+		}
 		if pod.DeletionTimestamp == nil {
 			podsToDelete = append(podsToDelete, pod)
 		}

kubernetes/test/e2e/pod_recycle_policy_test.go

@@ -57,7 +57,7 @@ var _ = Describe("Pod Recycle Policy", Ordered, func() {
 		By("patching controller deployment with restart-timeout for testing")
 		cmd = exec.Command("kubectl", "patch", "deployment", "opensandbox-controller-manager", "-n", namespace,
 			"--type", "json", "-p",
-			`[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value": "--restart-timeout=20s"}]`)
+			`[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value": "--restart-timeout=60s"}]`)
 		_, err = utils.Run(cmd)
 		Expect(err).NotTo(HaveOccurred(), "Failed to patch controller deployment")