Merge pull request #919 from Spground/feature/public-k8s-perf-opt

feat(k8s): Add some logs & perf opt

Author: Spground

kubernetes/Dockerfile

@@ -48,14 +48,16 @@ COPY internal/ internal/
 # by leaving it empty we can ensure that the container and binary shipped on it will have the same platform.
 RUN echo "Building for $TARGETOS/$TARGETARCH"
 ARG PACKAGE=./cmd/controller
+ARG COMMIT_ID=unknown
+ARG BUILD_DATE=unknown
 RUN if [ -n "${CC}" ]; then export CC; fi; \
     if [ -n "${CXX}" ]; then export CXX; fi; \
     export CGO_ENABLED="${CGO_ENABLED}" GOOS="${TARGETOS:-linux}" GOARCH="${TARGETARCH}" \
            CGO_CFLAGS="${CGO_CFLAGS:-${CFLAGS}}" \
            CGO_CXXFLAGS="${CGO_CXXFLAGS:-${CXXFLAGS}}" \
            CGO_LDFLAGS="${CGO_LDFLAGS}"; \
     go build ${GOFLAGS} -trimpath -buildvcs=false \
-    -ldflags "${LDFLAGS} -buildid= -B none" \
+    -ldflags "${LDFLAGS} -buildid= -B none -X main.commitID=${COMMIT_ID} -X main.buildDate=${BUILD_DATE}" \
     -o server ${PACKAGE}
 
 # Use golang image as base to ensure nsenter (util-linux) is available

kubernetes/build.sh

@@ -31,6 +31,8 @@ BUILD_ARGS=()
 for name in GOFLAGS LDFLAGS CGO_ENABLED CC CXX CFLAGS CXXFLAGS CGO_CFLAGS CGO_CXXFLAGS CGO_LDFLAGS; do
     build_arg_if_set "${name}"
 done
+BUILD_ARGS+=(--build-arg "COMMIT_ID=$(git rev-parse --short HEAD)")
+BUILD_ARGS+=(--build-arg "BUILD_DATE=$(date -u +%Y-%m-%dT%H:%M:%SZ)")
 mkdir -p "$(dirname "${BUILD_METADATA_FILE}")"
 
 DOCKERHUB_REPO="opensandbox"

kubernetes/cmd/controller/main.go

@@ -43,11 +43,17 @@ import (
 	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller"
 	poolassign "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller/poolassign"
 	cryptoutil "github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/crypto"
+	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/expectations"
 	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/fieldindex"
 	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/logging"
 	// +kubebuilder:scaffold:imports
 )
 
+var (
+	commitID  = "unknown"
+	buildDate = "unknown"
+)
+
 const (
 	defaultBatchSandboxConcurrency = 32
 	defaultPoolConcurrency         = 16
@@ -237,6 +243,8 @@ func main() {
 	logger := logging.NewLoggerWithZapOptions(logOpts)
 	ctrl.SetLogger(logger)
 
+	setupLog.Info("Starting controller", "commitID", commitID, "buildDate", buildDate)
+
 	// if the enable-http2 flag is false (the default), http/2 should be disabled
 	// due to its vulnerabilities. More specifically, disabling http/2 will
 	// prevent from being vulnerable to the HTTP/2 Stream Cancellation and
@@ -375,11 +383,12 @@ func main() {
 	}
 
 	config := ctrl.GetConfigOrDie()
+	config.UserAgent = "sandbox-k8s-controller/1.0"
 	// Set client rate limiter if specified
-	if kubeClientQPS > 0 {
+	if kubeClientQPS != 0 {
 		config.QPS = float32(kubeClientQPS)
 	}
-	if kubeClientBurst > 0 {
+	if kubeClientBurst != 0 {
 		config.Burst = kubeClientBurst
 	}
 
@@ -422,11 +431,12 @@ func main() {
 	}
 
 	if err := (&controller.BatchSandboxReconciler{
-		Client:           mgr.GetClient(),
-		Scheme:           mgr.GetScheme(),
-		Recorder:         mgr.GetEventRecorderFor("batchsandbox-controller"),
-		ResumePullSecret: resumePullSecret,
-		ProfileStore:     profileStore,
+		Client:              mgr.GetClient(),
+		Scheme:              mgr.GetScheme(),
+		Recorder:            mgr.GetEventRecorderFor("batchsandbox-controller"),
+		ResumePullSecret:    resumePullSecret,
+		ProfileStore:        profileStore,
+		StatusRVExpectation: expectations.NewResourceVersionExpectation(),
 	}).SetupWithManager(mgr, batchSandboxConcurrency); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "BatchSandbox")
 		os.Exit(1)

kubernetes/internal/controller/allocator.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -268,24 +269,38 @@ func NewAnnoAllocationSyncer(client client.Client) AllocationSyncer {
 }
 
 func (syncer *annoAllocationSyncer) SetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, allocation *SandboxAllocation) error {
-	old, ok := sandbox.DeepCopyObject().(*sandboxv1alpha1.BatchSandbox)
-	if !ok {
-		return fmt.Errorf("invalid object")
+	js, err := json.Marshal(allocation)
+	if err != nil {
+		return err
 	}
 	anno := sandbox.GetAnnotations()
 	if anno == nil {
 		anno = make(map[string]string)
 	}
-	js, err := json.Marshal(allocation)
+	anno[AnnoAllocStatusKey] = string(js)
+	sandbox.SetAnnotations(anno)
+
+	needAddFinalizer := !controllerutil.ContainsFinalizer(sandbox, FinalizerPoolAllocation)
+	if needAddFinalizer {
+		sandbox.SetFinalizers(append(sandbox.GetFinalizers(), FinalizerPoolAllocation))
+	}
+
+	meta := map[string]any{
+		"annotations": map[string]string{
+			AnnoAllocStatusKey: string(js),
+		},
+	}
+	if needAddFinalizer {
+		meta["finalizers"] = sandbox.GetFinalizers()
+	}
+	patchData, err := json.Marshal(map[string]any{"metadata": meta})
 	if err != nil {
 		return err
 	}
-	anno[AnnoAllocStatusKey] = string(js)
-	sandbox.SetAnnotations(anno)
-	// Add finalizer to ensure the sandbox is not deleted before all pods are recycled.
-	controllerutil.AddFinalizer(sandbox, FinalizerPoolAllocation)
-	patch := client.MergeFrom(old)
-	return syncer.client.Patch(ctx, sandbox, patch)
+	obj := &sandboxv1alpha1.BatchSandbox{}
+	obj.Name = sandbox.Name
+	obj.Namespace = sandbox.Namespace
+	return syncer.client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData))
 }
 
 func (syncer *annoAllocationSyncer) GetAllocation(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox) (*SandboxAllocation, error) {
@@ -340,20 +355,18 @@ func (syncer *annoAllocationSyncer) GetReleased(ctx context.Context, sandbox *sa
 }
 
 func (syncer *annoAllocationSyncer) SetReleased(ctx context.Context, sandbox *sandboxv1alpha1.BatchSandbox, released *AllocationReleased) error {
-	old, ok := sandbox.DeepCopyObject().(*sandboxv1alpha1.BatchSandbox)
-	if !ok {
-		return fmt.Errorf("invalid object")
+	js, err := json.Marshal(released)
+	if err != nil {
+		return err
 	}
 	anno := sandbox.GetAnnotations()
 	if anno == nil {
 		anno = make(map[string]string)
 	}
-	js, err := json.Marshal(released)
-	if err != nil {
-		return err
-	}
 	anno[AnnoAllocReleasedKey] = string(js)
 	sandbox.SetAnnotations(anno)
+
+	needRemoveFinalizer := false
 	// If the sandbox is being deleted and all allocated pods have been released,
 	// remove the finalizer so the sandbox can be garbage collected.
 	if !sandbox.DeletionTimestamp.IsZero() {
@@ -372,12 +385,34 @@ func (syncer *annoAllocationSyncer) SetReleased(ctx context.Context, sandbox *sa
 				break
 			}
 		}
-		if allReleased {
-			controllerutil.RemoveFinalizer(sandbox, FinalizerPoolAllocation)
+		if allReleased && controllerutil.ContainsFinalizer(sandbox, FinalizerPoolAllocation) {
+			needRemoveFinalizer = true
+			filtered := make([]string, 0, len(sandbox.GetFinalizers()))
+			for _, f := range sandbox.GetFinalizers() {
+				if f != FinalizerPoolAllocation {
+					filtered = append(filtered, f)
+				}
+			}
+			sandbox.SetFinalizers(filtered)
 		}
 	}
-	patch := client.MergeFrom(old)
-	return syncer.client.Patch(ctx, sandbox, patch)
+
+	meta := map[string]any{
+		"annotations": map[string]string{
+			AnnoAllocReleasedKey: string(js),
+		},
+	}
+	if needRemoveFinalizer {
+		meta["finalizers"] = sandbox.GetFinalizers()
+	}
+	patchData, err := json.Marshal(map[string]any{"metadata": meta})
+	if err != nil {
+		return err
+	}
+	obj := &sandboxv1alpha1.BatchSandbox{}
+	obj.Name = sandbox.Name
+	obj.Namespace = sandbox.Namespace
+	return syncer.client.Patch(ctx, obj, client.RawPatch(types.MergePatchType, patchData))
 }
 
 type AllocSpec struct {

kubernetes/internal/controller/batchsandbox_controller.go

@@ -66,10 +66,11 @@ type taskScheduleResult struct {
 // BatchSandboxReconciler reconciles a BatchSandbox object
 type BatchSandboxReconciler struct {
 	client.Client
-	Scheme         *runtime.Scheme
-	Recorder       record.EventRecorder
-	ProfileStore   *poolassign.ProfileStore
-	taskSchedulers sync.Map
+	Scheme              *runtime.Scheme
+	Recorder            record.EventRecorder
+	ProfileStore        *poolassign.ProfileStore
+	taskSchedulers      sync.Map
+	StatusRVExpectation expectations.ResourceVersionExpectation
 	// ResumePullSecret is the K8s Secret name for pulling snapshot images during resume.
 	ResumePullSecret string
 }
@@ -90,11 +91,13 @@ type BatchSandboxReconciler struct {
 //
 // For more details, check Reconcile and its Result here:
 // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.21.0/pkg/reconcile
-func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, retErr error) {
 	log := logf.FromContext(ctx)
+	start := time.Now()
 	var aggErrors []error
 	defer func() {
 		_ = DurationStore.Pop(req.String())
+		log.Info("Reconcile finished", "duration", time.Since(start).String(), "requeueAfter", result.RequeueAfter.String(), "error", retErr)
 	}()
 	batchSbx := &sandboxv1alpha1.BatchSandbox{}
 	if err := r.Get(ctx, client.ObjectKey{
@@ -192,6 +195,14 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	runtimeView := buildRuntimeView(batchSbx, pods)
+	// 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
+	// that must remain unacknowledged until resume completes and handlePause runs.
+	if batchSbx.Status.Phase != sandboxv1alpha1.BatchSandboxPhaseResuming &&
+		runtimeView.status.PauseObservedGeneration < batchSbx.Generation {
+		runtimeView.status.PauseObservedGeneration = batchSbx.Generation
+	}
 
 	if batchSbx.Status.Phase == sandboxv1alpha1.BatchSandboxPhasePaused {
 		r.deleteTaskScheduler(ctx, batchSbx)
@@ -210,9 +221,14 @@ func (r *BatchSandboxReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		}
 	}
 
-	aggErrors = append(aggErrors, r.persistRuntimeView(ctx, batchSbx, runtimeView)...)
+	requeue, persistErrors := r.persistRuntimeView(ctx, batchSbx, runtimeView)
+	aggErrors = append(aggErrors, persistErrors...)
 
-	return reconcile.Result{RequeueAfter: DurationStore.Pop(req.String())}, gerrors.Join(aggErrors...)
+	requeueAfter := DurationStore.Pop(req.String())
+	if requeue > 0 && (requeueAfter == 0 || requeue < requeueAfter) {
+		requeueAfter = requeue
+	}
+	return reconcile.Result{RequeueAfter: requeueAfter}, gerrors.Join(aggErrors...)
 }
 
 func calPodIndex(poolStrategy strategy.PoolStrategy, batchSbx *sandboxv1alpha1.BatchSandbox, pods []*corev1.Pod) (map[string]int, error) {

kubernetes/internal/controller/batchsandbox_pause_resume.go

@@ -209,10 +209,9 @@ func (r *BatchSandboxReconciler) dispatchPauseResume(ctx context.Context, bs *sa
 			result, err := r.handleResume(ctx, bs)
 			return result, true, err
 		}
-		log.Info("Dispatch: ACK only", "generation", generation, "pauseObservedGeneration", pauseObservedGen)
-		if err := r.ackPauseGeneration(ctx, bs); err != nil {
-			return ctrl.Result{}, true, err
-		}
+		// No pause intent — skip the dedicated ACK API call. The normal flow's
+		// persistRuntimeView will update PauseObservedGeneration in its status patch.
+		log.Info("Dispatch: no pause intent, deferring ACK to status patch", "generation", generation, "pauseObservedGeneration", pauseObservedGen)
 		return ctrl.Result{}, false, nil
 	}
 
@@ -479,8 +478,9 @@ func (r *BatchSandboxReconciler) completePause(ctx context.Context, bs *sandboxv
 
 	r.deleteTaskScheduler(ctx, bs)
 
+	var latest *sandboxv1alpha1.BatchSandbox
 	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		latest := &sandboxv1alpha1.BatchSandbox{}
+		latest = &sandboxv1alpha1.BatchSandbox{}
 		if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil {
 			return err
 		}
@@ -496,6 +496,7 @@ func (r *BatchSandboxReconciler) completePause(ctx context.Context, bs *sandboxv
 	}); err != nil {
 		return err
 	}
+	r.StatusRVExpectation.Expect(latest)
 
 	return nil
 }
@@ -584,8 +585,9 @@ func (r *BatchSandboxReconciler) continueResume(ctx context.Context, bs *sandbox
 }
 
 func (r *BatchSandboxReconciler) ackPauseGeneration(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox) error {
+	var latest *sandboxv1alpha1.BatchSandbox
 	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		latest := &sandboxv1alpha1.BatchSandbox{}
+		latest = &sandboxv1alpha1.BatchSandbox{}
 		if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil {
 			return err
 		}
@@ -595,14 +597,16 @@ func (r *BatchSandboxReconciler) ackPauseGeneration(ctx context.Context, bs *san
 	}); err != nil {
 		return err
 	}
+	r.StatusRVExpectation.Expect(latest)
 	bs.Status.PauseObservedGeneration = bs.Generation
 	applyBatchSandboxPhaseConditions(&bs.Status)
 	return nil
 }
 
 func (r *BatchSandboxReconciler) ackPauseWithPhase(ctx context.Context, bs *sandboxv1alpha1.BatchSandbox, phase sandboxv1alpha1.BatchSandboxPhase, _ string) error {
+	var latest *sandboxv1alpha1.BatchSandbox
 	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		latest := &sandboxv1alpha1.BatchSandbox{}
+		latest = &sandboxv1alpha1.BatchSandbox{}
 		if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil {
 			return err
 		}
@@ -613,6 +617,7 @@ func (r *BatchSandboxReconciler) ackPauseWithPhase(ctx context.Context, bs *sand
 	}); err != nil {
 		return err
 	}
+	r.StatusRVExpectation.Expect(latest)
 	bs.Status.PauseObservedGeneration = bs.Generation
 	bs.Status.Phase = phase
 	applyBatchSandboxPhaseConditions(&bs.Status)
@@ -640,8 +645,9 @@ func (r *BatchSandboxReconciler) setCondition(
 	reason string,
 	message string,
 ) error {
-	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		latest := &sandboxv1alpha1.BatchSandbox{}
+	var latest *sandboxv1alpha1.BatchSandbox
+	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		latest = &sandboxv1alpha1.BatchSandbox{}
 		if err := r.Get(ctx, types.NamespacedName{Namespace: bs.Namespace, Name: bs.Name}, latest); err != nil {
 			return err
 		}
@@ -674,5 +680,9 @@ func (r *BatchSandboxReconciler) setCondition(
 
 		latest.Status.Conditions = conditions
 		return r.Status().Update(ctx, latest)
-	})
+	}); err != nil {
+		return err
+	}
+	r.StatusRVExpectation.Expect(latest)
+	return nil
 }