feat(k8s): add batch sandbox expire & endpoints status

Spground · Spground · commit c403cfa7dae1 · 2025-12-18T18:24:53.000+08:00
diff --git a/kubernetes/config/samples/sandbox_v1alpha1_batchsandbox-with-task.yaml b/kubernetes/config/samples/sandbox_v1alpha1_batchsandbox-with-task.yaml
@@ -0,0 +1,47 @@
+apiVersion: sandbox.opensandbox.io/v1alpha1
+kind: BatchSandbox
+metadata:
+  labels:
+    app.kubernetes.io/name: sandbox-k8s
+    app.kubernetes.io/managed-by: kustomize
+  name: batchsandbox-sample
+  namespace: sandbox-k8s
+spec:
+  replicas: 2
+  template:
+    metadata:
+      labels:
+        app: example
+    spec:
+      containers:
+      - name: main
+        image: registry.k8s.io/e2e-test-images/httpd:2.4.38-4
+        command:
+        - tail
+        - -f
+        - /dev/null
+  expireTime: "2025-12-03T12:55:41Z"
+  taskTemplate:
+    metadata:
+      labels:
+        app.task: task
+    spec:
+      container: # container mode
+        image: example.com/agent-task:latest
+        command:
+        - /usr/bin/sleep
+        args:
+        - "1"
+  shardTaskPatches: # patch list for container tasks
+  - spec:
+      container:
+        command: # patch command and args, the final command is `python -m http.server 8080`
+        - python
+        args:
+        - -m
+        - http.server
+        - "8080"
+  - spec:
+      container:
+        args: # patch args only, the final command is `/usr/bin/sleep 456`
+        - "456"
diff --git a/kubernetes/config/samples/sandbox_v1alpha1_batchsandbox.yaml b/kubernetes/config/samples/sandbox_v1alpha1_batchsandbox.yaml
@@ -20,28 +20,3 @@ spec:
         - tail
         - -f
         - /dev/null
-  expireTime: "2025-12-03T12:55:41Z"
-  taskTemplate:
-    metadata:
-      labels:
-        app.task: task
-    spec:
-      container: # container mode
-        image: example.com/agent-task:latest
-        command:
-        - /usr/bin/sleep
-        args:
-        - "1"
-  shardTaskPatches: # patch list for container tasks
-  - spec:
-      container:
-        command: # patch command and args, the final command is `python -m http.server 8080`
-        - python
-        args:
-        - -m
-        - http.server
-        - "8080"
-  - spec:
-      container:
-        args: # patch args only, the final command is `/usr/bin/sleep 456`
-        - "456"
diff --git a/kubernetes/config/samples/sandbox_v1alpha1_pool.yaml b/kubernetes/config/samples/sandbox_v1alpha1_pool.yaml
@@ -13,10 +13,12 @@ spec:
         app: example
     spec:
       containers:
-        - name: executor
-          image: reg.docker.alibaba-inc.com/sandbox-k8s/task-executor:1.0.6
+        - name: main
+          image: registry.k8s.io/e2e-test-images/httpd:2.4.38-4
           command:
-            - /server
+          - tail
+          - -f
+          - /dev/null
       tolerations:
         - operator: "Exists"
   capacitySpec:
diff --git a/kubernetes/config/samples/sandbox_v1alpha1_pooled_batchsandbox.yaml b/kubernetes/config/samples/sandbox_v1alpha1_pooled_batchsandbox.yaml
@@ -9,4 +9,4 @@ metadata:
 spec:
   poolRef: pool-sample
   replicas: 2
-  expireTime: "2025-12-03T12:55:41Z"
+  expireTime: "2026-12-03T12:55:41Z"
diff --git a/kubernetes/internal/controller/apis.go b/kubernetes/internal/controller/apis.go
@@ -30,6 +30,8 @@ const (
 	AnnoPoolAllocGenerationKey = "pool.opensandbox.io/alloc-generation"
 
 	FinalizerTaskCleanup = "batch-sandbox.sandbox.opensandbox.io/task-cleanup"
+
+	AnnotationSandboxEndpoints = "sandbox.opensandbox.io/endpoints"
 )
 
 type SandboxAllocation struct {
diff --git a/kubernetes/internal/controller/batchsandbox_controller.go b/kubernetes/internal/controller/batchsandbox_controller.go
@@ -112,6 +112,23 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		}
 		return ctrl.Result{}, err
 	}
+	// handle expire
+	if expireAt := batchSbx.Spec.ExpireTime; expireAt != nil {
+		now := time.Now()
+		if expireAt.Time.Before(now) {
+			if batchSbx.DeletionTimestamp == nil {
+				klog.Infof("batch sandbox %s expired, expire at %v, delete", klog.KObj(batchSbx), expireAt)
+				if err := r.Delete(ctx, batchSbx); err != nil {
+					if errors.IsNotFound(err) {
+						return ctrl.Result{}, nil
+					}
+					return ctrl.Result{}, err
+				}
+			}
+		} else {
+			DurationStore.Push(types.NamespacedName{Namespace: batchSbx.Namespace, Name: batchSbx.Name}.String(), expireAt.Time.Sub(now))
+		}
+	}
 
 	// handle finalizers
 	if batchSbx.DeletionTimestamp == nil {
@@ -153,15 +170,32 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		newStatus.Replicas = 0
 		newStatus.Allocated = 0
 		newStatus.Ready = 0
+		ipList := []string{}
 		for _, pod := range pods {
 			newStatus.Replicas++
 			if utils.IsAssigned(pod) {
 				newStatus.Allocated++
+				ipList = append(ipList, pod.Status.PodIP)
 			}
 			if pod.Status.Phase == corev1.PodRunning && utils.IsPodReady(pod) {
 				newStatus.Ready++
 			}
 		}
+		raw, _ := json.Marshal(ipList)
+		if batchSbx.Annotations[AnnotationSandboxEndpoints] != string(raw) {
+			patchData, _ := json.Marshal(map[string]interface{}{
+				"metadata": map[string]interface{}{
+					"annotations": map[string]string{
+						AnnotationSandboxEndpoints: string(raw),
+					},
+				},
+			})
+			obj := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Namespace: batchSbx.Namespace, Name: batchSbx.Name}}
+			if err := r.Patch(ctx, obj, client.RawPatch(types.MergePatchType, []byte(patchData))); err != nil {
+				klog.Errorf("failed to patch annotation %s, %s, body %s", AnnotationSandboxEndpoints, klog.KObj(batchSbx), patchData)
+				errors = append(errors, err)
+			}
+		}
 		if !reflect.DeepEqual(newStatus, batchSbx.Status) {
 			klog.Infof("To update BatchSandbox status for %s, replicas=%d allocated=%d ready=%d", klog.KObj(batchSbx), newStatus.Replicas, newStatus.Allocated, newStatus.Ready)
 			if err := r.updateStatus(batchSbx, newStatus); err != nil {
@@ -417,7 +451,7 @@ func (r *BatchSandboxReconciler) releasePods(ctx context.Context, batchSbx *sand
 			Name:      batchSbx.Name,
 		},
 	}
-	return r.Client.Patch(ctx, b, client.RawPatch(types.StrategicMergePatchType, []byte(body)))
+	return r.Client.Patch(ctx, b, client.RawPatch(types.MergePatchType, []byte(body)))
 }
 
 // Normal Mode
diff --git a/kubernetes/internal/controller/batchsandbox_controller_test.go b/kubernetes/internal/controller/batchsandbox_controller_test.go
@@ -32,8 +32,10 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
 	gerrors "errors"
 	"fmt"
+	"net"
 	"reflect"
 	"sync"
 	"testing"
@@ -57,7 +59,9 @@ import (
 	"k8s.io/apimachinery/pkg/util/rand"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/utils/ptr"
+	"k8s.io/utils/set"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -122,18 +126,49 @@ var _ = Describe("BatchSandbox Controller", func() {
 			err := k8sClient.Get(ctx, typeNamespacedName, resource)
 			if !errors.IsNotFound(err) {
 				Expect(err).NotTo(HaveOccurred())
+			} else {
+				return
 			}
 			By(fmt.Sprintf("Cleanup the specific resource instance BatchSandbox %s", typeNamespacedName))
 			Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
 		})
-		It("should successfully update batch sandbox status", func() {
+		It("should successfully create pod, update batch sandbox status, endpoints info", func() {
+			wantIPSet := make(set.Set[string])
 			Eventually(func(g Gomega) {
 				bs := &sandboxv1alpha1.BatchSandbox{}
 				if err := k8sClient.Get(ctx, typeNamespacedName, bs); err != nil {
 					return
 				}
+				allPods := &corev1.PodList{}
+				g.Expect(k8sClient.List(ctx, allPods, &client.ListOptions{Namespace: bs.Namespace})).Should(Succeed())
+				pods := []*corev1.Pod{}
+				for i := range allPods.Items {
+					po := &allPods.Items[i]
+					if metav1.IsControlledBy(po, bs) {
+						pods = append(pods, po)
+						if po.Status.PodIP != "" {
+							continue
+						}
+						// patch status to make pod Scheduled
+						mockIP := randomIPv4().String()
+						wantIPSet.Insert(mockIP)
+						po.Status.PodIP = mockIP
+						po.Status.Phase = corev1.PodRunning
+						po.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}
+						Expect(k8sClient.Status().Update(context.Background(), po)).To(Succeed())
+					}
+				}
+				g.Expect(len(pods)).To(Equal(int(*bs.Spec.Replicas)))
 				g.Expect(bs.Status.ObservedGeneration).To(Equal(bs.Generation))
 				g.Expect(bs.Status.Replicas).To(Equal(*bs.Spec.Replicas))
+				g.Expect(bs.Status.Allocated).To(Equal(*bs.Spec.Replicas))
+				g.Expect(bs.Status.Ready).To(Equal(*bs.Spec.Replicas))
+
+				gotIPs := []string{}
+				if raw := bs.Annotations[AnnotationSandboxEndpoints]; raw != "" {
+					json.Unmarshal([]byte(raw), &gotIPs)
+				}
+				g.Expect(wantIPSet.Equal(set.New(gotIPs...))).To(BeTrue(), fmt.Sprintf("wantIPSet %v, gotIPs %v", wantIPSet.SortedList(), gotIPs))
 			}, timeout, interval).Should(Succeed())
 		})
 		It("should successfully correctly create new Pod and update batch sandbox status when user scale out", func() {
@@ -190,6 +225,33 @@ var _ = Describe("BatchSandbox Controller", func() {
 				g.Expect(newPod.CreationTimestamp).NotTo(Equal(oldPod.CreationTimestamp))
 			}, timeout, interval).Should(Succeed())
 		})
+		It("should delete batch sandbox and related Pods for expired batch sandbox", func() {
+			Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {
+				bs := &sandboxv1alpha1.BatchSandbox{}
+				if err := k8sClient.Get(ctx, typeNamespacedName, bs); err != nil {
+					return err
+				}
+				bs.Spec.ExpireTime = &metav1.Time{Time: time.Now().Add(3 * time.Second)}
+				return k8sClient.Update(ctx, bs)
+			})).Should(Succeed())
+
+			Eventually(
+				func(g Gomega) {
+					bs := &sandboxv1alpha1.BatchSandbox{}
+					g.Expect(errors.IsNotFound(k8sClient.Get(ctx, typeNamespacedName, bs))).To(BeTrue())
+					allPods := &corev1.PodList{}
+					g.Expect(k8sClient.List(ctx, allPods, &client.ListOptions{Namespace: bs.Namespace})).Should(Succeed())
+					pods := []*corev1.Pod{}
+					for i := range allPods.Items {
+						po := &allPods.Items[i]
+						if metav1.IsControlledBy(po, bs) {
+							pods = append(pods, po)
+						}
+					}
+					g.Expect(len(pods)).To(BeZero())
+				},
+				timeout, interval).Should(Succeed())
+		})
 	})
 
 	// Pooling Mode
@@ -233,7 +295,7 @@ var _ = Describe("BatchSandbox Controller", func() {
 			Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
 		})
 
-		It("should successfully update batch sandbox status when get pod from pool alloc", func() {
+		It("should successfully update batch sandbox status, sbx endpoints info when get pod from pool alloc", func() {
 			// mock pool allocation
 			mockPods := []string{}
 			for i := range replicas {
@@ -256,10 +318,14 @@ var _ = Describe("BatchSandbox Controller", func() {
 				po.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: corev1.ConditionTrue}}
 				Expect(k8sClient.Status().Update(context.Background(), po)).To(Succeed())
 			}
-			bs := &sandboxv1alpha1.BatchSandbox{}
-			Expect(k8sClient.Get(ctx, typeNamespacedName, bs)).To(Succeed())
-			setSandboxAllocation(bs, SandboxAllocation{Pods: mockPods})
-			Expect(k8sClient.Update(ctx, bs)).To(Succeed())
+			Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {
+				bs := &sandboxv1alpha1.BatchSandbox{}
+				if err := k8sClient.Get(ctx, typeNamespacedName, bs); err != nil {
+					return err
+				}
+				setSandboxAllocation(bs, SandboxAllocation{Pods: mockPods})
+				return k8sClient.Update(ctx, bs)
+			})).Should(Succeed())
 			By(fmt.Sprintf("Mock pool allocate Pod %v for BatchSandbox %s", mockPods, typeNamespacedName))
 
 			Eventually(func(g Gomega) {
@@ -271,11 +337,22 @@ var _ = Describe("BatchSandbox Controller", func() {
 				g.Expect(bs.Status.Replicas).To(Equal(*bs.Spec.Replicas))
 				g.Expect(bs.Status.Allocated).To(Equal(*bs.Spec.Replicas))
 				g.Expect(bs.Status.Ready).To(Equal(*bs.Spec.Replicas))
+
+				g.Expect(bs.Annotations[AnnotationSandboxEndpoints]).To(Equal("[\"1.2.3.4\"]"))
 			}, timeout, interval).Should(Succeed())
 		})
 	})
 })
 
+func randomIPv4() net.IP {
+	rand.Seed(time.Now().UnixNano())
+	ip := make(net.IP, 4)
+	for i := range ip {
+		ip[i] = byte(rand.Intn(256))
+	}
+	return ip
+}
+
 var _ = Describe("BatchSandbox Task Scheduler", func() {
 	var (
 		timeout  = 30 * time.Second
diff --git a/kubernetes/internal/controller/pool_controller_test.go b/kubernetes/internal/controller/pool_controller_test.go
@@ -44,6 +44,7 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/rand"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/utils/ptr"
 	kclient "sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -275,13 +276,20 @@ var _ = Describe("Pool update", func() {
 			Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
 		})
 		It("should successfully update pool revision", func() {
-			pool := &sandboxv1alpha1.Pool{}
-			Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
-			oldRevision := pool.Status.Revision
-			pool.Spec.Template.Labels = map[string]string{
-				"test.pool.update": "v1",
-			}
-			Expect(k8sClient.Update(ctx, pool)).To(Succeed())
+			var oldRevision string
+			Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {
+				pool := &sandboxv1alpha1.Pool{}
+				if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
+					return err
+				}
+				if oldRevision == "" {
+					oldRevision = pool.Status.Revision
+				}
+				pool.Spec.Template.Labels = map[string]string{
+					"test.pool.update": "v1",
+				}
+				return k8sClient.Update(ctx, pool)
+			})).Should(Succeed())
 			Eventually(func(g Gomega) {
 				pool := &sandboxv1alpha1.Pool{}
 				Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
diff --git a/kubernetes/internal/utils/pod.go b/kubernetes/internal/utils/pod.go
@@ -169,7 +169,7 @@ func getPodsPrefix(controllerName string) string {
 }
 
 func IsAssigned(pod *v1.Pod) bool {
-	return pod != nil && pod.Spec.NodeName != "" && pod.Status.PodIP != ""
+	return pod != nil && (pod.Spec.NodeName != "" || pod.Status.PodIP != "")
 }
 
 func PodNameSorter(a, b *v1.Pod) int {

Original file line number	Diff line number	Diff line change
`@@ -30,6 +30,8 @@ const (`
`30`	`30`	`AnnoPoolAllocGenerationKey = "pool.opensandbox.io/alloc-generation"`
`31`	`31`
`32`	`32`	`FinalizerTaskCleanup = "batch-sandbox.sandbox.opensandbox.io/task-cleanup"`
	`33`	`+`
	`34`	`+ AnnotationSandboxEndpoints = "sandbox.opensandbox.io/endpoints"`
`33`	`35`	`)`
`34`	`36`
`35`	`37`	`type SandboxAllocation struct {`
Original file line number	Diff line number	Diff line change
`@@ -169,7 +169,7 @@ func getPodsPrefix(controllerName string) string {`
`169`	`169`	`}`
`170`	`170`
`171`	`171`	`func IsAssigned(pod *v1.Pod) bool {`
`172`		`- return pod != nil && pod.Spec.NodeName != "" && pod.Status.PodIP != ""`
	`172`	`+ return pod != nil && (pod.Spec.NodeName != "" \|\| pod.Status.PodIP != "")`
`173`	`173`	`}`
`174`	`174`
`175`	`175`	`func PodNameSorter(a, b *v1.Pod) int {`