diff --git a/docs/kubernetes/index.md b/docs/kubernetes/index.md index a1104c65a..ff0a48942 100644 --- a/docs/kubernetes/index.md +++ b/docs/kubernetes/index.md @@ -34,6 +34,13 @@ The Pool custom resource maintains a pool of pre-warmed compute resources to ena - Automatic resource allocation and deallocation based on demand - Real-time status monitoring showing total, allocated, and available resources +When a warm Pod is allocated, the controller labels it with +`batch-sandbox.sandbox.opensandbox.io/name=` before it +publishes the allocation annotation. The same label is already present on +non-pooled Pods. Operators and node-local telemetry can therefore use one stable +BatchSandbox identity in either lifecycle mode. The controller removes the +label when a warm Pod returns to the idle pool, before it can be reused. + ### Pod Eviction Pool supports graceful pod eviction for scenarios like node maintenance or resource reclamation: diff --git a/kubernetes/internal/controller/pool_controller.go b/kubernetes/internal/controller/pool_controller.go index 6856fb6bc..2a30328f9 100644 --- a/kubernetes/internal/controller/pool_controller.go +++ b/kubernetes/internal/controller/pool_controller.go @@ -535,8 +535,53 @@ func (r *PoolReconciler) doAllocate(ctx context.Context, pool *sandboxv1alpha1.P // 1. Compute latest allocated pods per sandbox (merge current + newly allocated). toSyncMap := r.getLatestAllocated(ctx, pool, batchSandboxes, toAllocate) - // 2. Concurrently sync each sandbox's Allocated annotation (AddFinalizer is called inside SyncSandboxAllocation). - return r.syncSandboxConcurrently(ctx, batchSandboxes, toSyncMap, r.Allocator.SyncSandboxAllocation, "allocated") + // 2. Label newly allocated pods before publishing them through the sandbox allocation annotation. + // BatchSandbox reconciliation can start work as soon as that annotation is visible. + newPodAllocation := make(map[string]string) + for sandboxName, podNames := range toAllocate { + if _, willSync := toSyncMap[sandboxName]; !willSync { + continue + } + for _, podName := range podNames { + newPodAllocation[podName] = sandboxName + } + } + newlyAllocatedPods := make([]*corev1.Pod, 0, len(newPodAllocation)) + missingPods := make(map[string]struct{}, len(newPodAllocation)) + for podName := range newPodAllocation { + missingPods[podName] = struct{}{} + } + for _, pod := range pods { + if _, newlyAllocated := newPodAllocation[pod.Name]; newlyAllocated { + newlyAllocatedPods = append(newlyAllocatedPods, pod) + delete(missingPods, pod.Name) + } + } + if len(missingPods) > 0 { + missingPodNames := make([]string, 0, len(missingPods)) + for podName := range missingPods { + missingPodNames = append(missingPodNames, podName) + } + sort.Strings(missingPodNames) + return fmt.Errorf("newly allocated pool pods not found: %v", missingPodNames) + } + if err := r.syncPodSandboxLabels(ctx, newlyAllocatedPods, newPodAllocation); err != nil { + return err + } + + // 3. Concurrently sync each sandbox's Allocated annotation (AddFinalizer is called inside SyncSandboxAllocation). + if err := r.syncSandboxConcurrently(ctx, batchSandboxes, toSyncMap, r.Allocator.SyncSandboxAllocation, "allocated"); err != nil { + // SyncSandboxAllocation rolls its in-memory entry back when annotation + // publication fails. Converge labels to that final store view so a Pod + // never retains identity for an allocation that was not published, while + // preserving labels for concurrently successful publications. + finalAllocation, getErr := r.Allocator.GetPoolAllocation(ctx, pool) + if getErr != nil { + return gerrors.Join(err, fmt.Errorf("failed to get allocation for label rollback: %w", getErr)) + } + return gerrors.Join(err, r.syncPodSandboxLabels(ctx, pods, finalAllocation)) + } + return nil } // getLatestAllocated computes the latest allocated pods for each sandbox by merging current allocation with new pods to allocate. @@ -810,6 +855,9 @@ func (r *PoolReconciler) scheduleSandbox(ctx context.Context, pool *sandboxv1alp if err != nil { return nil, err } + if err := r.syncPodSandboxLabels(ctx, pods, latestAllocation); err != nil { + return nil, err + } idlePods := make([]string, 0) for _, pod := range pods { if _, ok := latestAllocation[pod.Name]; !ok { @@ -827,6 +875,61 @@ func (r *PoolReconciler) scheduleSandbox(ctx context.Context, pool *sandboxv1alp return result, nil } +// syncPodSandboxLabels converges pool pod identity labels to the allocator's +// final allocation view without changing allocation or scheduling decisions. +func (r *PoolReconciler) syncPodSandboxLabels(ctx context.Context, pods []*corev1.Pod, podAllocation map[string]string) error { + errCh := make(chan error, len(pods)) + sem := make(chan struct{}, syncSandboxAllocConcurrency) + var wg sync.WaitGroup + + for _, pod := range pods { + sandboxName, allocated := podAllocation[pod.Name] + currentSandbox, hasLabel := pod.Labels[LabelBatchSandboxNameKey] + if (allocated && hasLabel && currentSandbox == sandboxName) || (!allocated && !hasLabel) { + continue + } + + // Acquire before starting the goroutine so a large reconciliation does + // not create one blocked goroutine per Pod. + sem <- struct{}{} + wg.Add(1) + go func(pod *corev1.Pod, sandboxName string, allocated bool) { + defer wg.Done() + defer func() { <-sem }() + + updated := pod.DeepCopy() + if allocated { + if updated.Labels == nil { + updated.Labels = make(map[string]string) + } + updated.Labels[LabelBatchSandboxNameKey] = sandboxName + } else { + delete(updated.Labels, LabelBatchSandboxNameKey) + } + if err := r.Patch(ctx, updated, client.MergeFrom(pod)); err != nil { + // A disappearing idle/released Pod needs no cleanup. A newly + // allocated Pod disappearing before its identity is written must + // stop allocation publication. + if !errors.IsNotFound(err) || allocated { + errCh <- fmt.Errorf("failed to sync sandbox label on pod %s: %w", pod.Name, err) + } + return + } + // Keep this reconcile's list snapshot current so the final convergence + // pass does not repeat the same patch. + pod.Labels = updated.Labels + }(pod, sandboxName, allocated) + } + + wg.Wait() + close(errCh) + var errs []error + for err := range errCh { + errs = append(errs, err) + } + return gerrors.Join(errs...) +} + func (r *PoolReconciler) updatePool(ctx context.Context, pool *sandboxv1alpha1.Pool, pods []*corev1.Pod, idlePods []string) (*UpdateResult, error) { updateRevision, err := r.calculateRevision(pool) if err != nil { diff --git a/kubernetes/internal/controller/pool_sandbox_label_test.go b/kubernetes/internal/controller/pool_sandbox_label_test.go new file mode 100644 index 000000000..327465462 --- /dev/null +++ b/kubernetes/internal/controller/pool_sandbox_label_test.go @@ -0,0 +1,329 @@ +// Copyright 2025 Alibaba Group Holding Ltd. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "context" + "errors" + "fmt" + goruntime "runtime" + "sync" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/golang/mock/gomock" + "github.com/stretchr/testify/require" + + sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1" +) + +func TestDoAllocateLabelsPodBeforeAllocationSync(t *testing.T) { + ctx := context.Background() + pod := testPoolPod("pool-pod", map[string]string{LabelPoolName: "pool"}) + sandbox := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: pod.Namespace}} + reconciler, countingClient := newPodLabelTestReconciler(t, pod) + order := &labelSyncOrder{} + countingClient.onPodPatch = func() { order.record("label-patch") } + + mockController := gomock.NewController(t) + allocator := NewMockAllocator(mockController) + allocator.EXPECT().GetSandboxAllocation(gomock.Any(), sandbox).Return([]string{}, nil) + allocator.EXPECT().SyncSandboxAllocation(gomock.Any(), sandbox, []string{pod.Name}).DoAndReturn( + func(ctx context.Context, _ *sandboxv1alpha1.BatchSandbox, _ []string) error { + updated := &corev1.Pod{} + if err := reconciler.Get(ctx, types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, updated); err != nil { + return err + } + if got := updated.Labels[LabelBatchSandboxNameKey]; got != sandbox.Name { + return fmt.Errorf("sandbox label = %q, want %q", got, sandbox.Name) + } + order.record("allocation-sync") + return nil + }, + ) + reconciler.Allocator = allocator + + require.NoError(t, reconciler.doAllocate( + ctx, + &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool", Namespace: pod.Namespace}}, + []*sandboxv1alpha1.BatchSandbox{sandbox}, + []*corev1.Pod{pod.DeepCopy()}, + map[string][]string{sandbox.Name: {pod.Name}}, + )) + require.Equal(t, []string{"label-patch", "allocation-sync"}, order.events()) +} + +func TestDoAllocateDoesNotPublishMissingPod(t *testing.T) { + ctx := context.Background() + pod := testPoolPod("missing-pod", map[string]string{LabelPoolName: "pool"}) + sandbox := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: pod.Namespace}} + reconciler, countingClient := newPodLabelTestReconciler(t, pod) + countingClient.podPatchErr = apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, pod.Name) + + mockController := gomock.NewController(t) + allocator := NewMockAllocator(mockController) + allocator.EXPECT().GetSandboxAllocation(gomock.Any(), sandbox).Return([]string{}, nil) + reconciler.Allocator = allocator + + err := reconciler.doAllocate( + ctx, + &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool", Namespace: pod.Namespace}}, + []*sandboxv1alpha1.BatchSandbox{sandbox}, + []*corev1.Pod{pod.DeepCopy()}, + map[string][]string{sandbox.Name: {pod.Name}}, + ) + require.ErrorIs(t, err, countingClient.podPatchErr) +} + +func TestDoAllocateRollsBackOnlyFailedPublicationLabels(t *testing.T) { + ctx := context.Background() + podA := testPoolPod("pool-pod-a", map[string]string{LabelPoolName: "pool"}) + podB := testPoolPod("pool-pod-b", map[string]string{LabelPoolName: "pool"}) + sandboxA := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: podA.Namespace}} + sandboxB := &sandboxv1alpha1.BatchSandbox{ObjectMeta: metav1.ObjectMeta{Name: "sandbox-b", Namespace: podB.Namespace}} + pool := &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool", Namespace: podA.Namespace}} + reconciler, _ := newPodLabelTestReconciler(t, podA, podB) + publicationErr := errors.New("allocation publication failed") + + mockController := gomock.NewController(t) + allocator := NewMockAllocator(mockController) + allocator.EXPECT().GetSandboxAllocation(gomock.Any(), sandboxA).Return([]string{}, nil) + allocator.EXPECT().GetSandboxAllocation(gomock.Any(), sandboxB).Return([]string{}, nil) + allocator.EXPECT().SyncSandboxAllocation(gomock.Any(), sandboxA, []string{podA.Name}).Return(nil) + allocator.EXPECT().SyncSandboxAllocation(gomock.Any(), sandboxB, []string{podB.Name}).Return(publicationErr) + allocator.EXPECT().GetPoolAllocation(gomock.Any(), pool).Return(map[string]string{podA.Name: sandboxA.Name}, nil) + reconciler.Allocator = allocator + + err := reconciler.doAllocate( + ctx, + pool, + []*sandboxv1alpha1.BatchSandbox{sandboxA, sandboxB}, + []*corev1.Pod{podA.DeepCopy(), podB.DeepCopy()}, + map[string][]string{sandboxA.Name: {podA.Name}, sandboxB.Name: {podB.Name}}, + ) + require.ErrorIs(t, err, publicationErr) + + updatedA := getPodLabelTestPod(t, ctx, reconciler.Client, podA) + updatedB := getPodLabelTestPod(t, ctx, reconciler.Client, podB) + require.Equal(t, sandboxA.Name, updatedA.Labels[LabelBatchSandboxNameKey]) + _, hasFailedSandboxLabel := updatedB.Labels[LabelBatchSandboxNameKey] + require.False(t, hasFailedSandboxLabel) +} + +func TestSyncPodSandboxLabelsAllocation(t *testing.T) { + ctx := context.Background() + pod := testPoolPod("pool-pod", map[string]string{LabelPoolName: "pool"}) + reconciler, countingClient := newPodLabelTestReconciler(t, pod) + listedPod := pod.DeepCopy() + + require.NoError(t, reconciler.syncPodSandboxLabels(ctx, []*corev1.Pod{listedPod}, map[string]string{pod.Name: "sandbox-a"})) + require.NoError(t, reconciler.syncPodSandboxLabels(ctx, []*corev1.Pod{listedPod}, map[string]string{pod.Name: "sandbox-a"})) + + updated := getPodLabelTestPod(t, ctx, reconciler.Client, pod) + require.Equal(t, "sandbox-a", updated.Labels[LabelBatchSandboxNameKey]) + require.Equal(t, "pool", updated.Labels[LabelPoolName]) + require.Equal(t, 1, countingClient.podPatchCalls()) +} + +func TestSyncPodSandboxLabelsRelease(t *testing.T) { + ctx := context.Background() + pod := testPoolPod("pool-pod", map[string]string{ + LabelPoolName: "pool", + LabelBatchSandboxNameKey: "sandbox-a", + }) + reconciler, countingClient := newPodLabelTestReconciler(t, pod) + + require.NoError(t, reconciler.syncPodSandboxLabels(ctx, []*corev1.Pod{pod.DeepCopy()}, map[string]string{})) + + updated := getPodLabelTestPod(t, ctx, reconciler.Client, pod) + _, hasSandboxLabel := updated.Labels[LabelBatchSandboxNameKey] + require.False(t, hasSandboxLabel) + require.Equal(t, "pool", updated.Labels[LabelPoolName]) + require.Equal(t, 1, countingClient.podPatchCalls()) +} + +func TestSyncPodSandboxLabelsReassignment(t *testing.T) { + ctx := context.Background() + pod := testPoolPod("pool-pod", map[string]string{ + LabelPoolName: "pool", + LabelBatchSandboxNameKey: "sandbox-a", + }) + reconciler, countingClient := newPodLabelTestReconciler(t, pod) + + require.NoError(t, reconciler.syncPodSandboxLabels(ctx, []*corev1.Pod{pod.DeepCopy()}, map[string]string{pod.Name: "sandbox-b"})) + + updated := getPodLabelTestPod(t, ctx, reconciler.Client, pod) + require.Equal(t, "sandbox-b", updated.Labels[LabelBatchSandboxNameKey]) + require.Equal(t, 1, countingClient.podPatchCalls()) +} + +func TestSyncPodSandboxLabelsNoOp(t *testing.T) { + ctx := context.Background() + allocatedPod := testPoolPod("allocated-pod", map[string]string{ + LabelPoolName: "pool", + LabelBatchSandboxNameKey: "sandbox-a", + }) + idlePod := testPoolPod("idle-pod", map[string]string{LabelPoolName: "pool"}) + reconciler, countingClient := newPodLabelTestReconciler(t, allocatedPod, idlePod) + + require.NoError(t, reconciler.syncPodSandboxLabels( + ctx, + []*corev1.Pod{allocatedPod.DeepCopy(), idlePod.DeepCopy()}, + map[string]string{allocatedPod.Name: "sandbox-a"}, + )) + + require.Equal(t, 0, countingClient.podPatchCalls()) +} + +func TestSyncPodSandboxLabelsBoundsWaitingGoroutines(t *testing.T) { + const podCount = 200 + originalConcurrency := syncSandboxAllocConcurrency + syncSandboxAllocConcurrency = 8 + t.Cleanup(func() { syncSandboxAllocConcurrency = originalConcurrency }) + + ctx := context.Background() + pods := make([]*corev1.Pod, 0, podCount) + objects := make([]client.Object, 0, podCount) + allocation := make(map[string]string, podCount) + for i := 0; i < podCount; i++ { + pod := testPoolPod(fmt.Sprintf("pool-pod-%03d", i), map[string]string{LabelPoolName: "pool"}) + pods = append(pods, pod.DeepCopy()) + objects = append(objects, pod) + allocation[pod.Name] = fmt.Sprintf("sandbox-%03d", i) + } + + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + blockingClient := &blockingPodPatchClient{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build(), + started: make(chan struct{}, podCount), + release: make(chan struct{}), + } + reconciler := &PoolReconciler{Client: blockingClient} + done := make(chan error, 1) + baseline := goruntime.NumGoroutine() + go func() { + done <- reconciler.syncPodSandboxLabels(ctx, pods, allocation) + }() + + for i := 0; i < syncSandboxAllocConcurrency; i++ { + select { + case <-blockingClient.started: + case <-time.After(time.Second): + close(blockingClient.release) + require.FailNow(t, "timed out waiting for bounded patch workers") + } + } + waitingGoroutines := goruntime.NumGoroutine() - baseline + close(blockingClient.release) + require.NoError(t, <-done) + require.Less(t, waitingGoroutines, 50, "label sync should not create one waiting goroutine per pod") +} + +func testPoolPod(name string, labels map[string]string) *corev1.Pod { + return &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default", Labels: labels}} +} + +func newPodLabelTestReconciler(t *testing.T, pods ...*corev1.Pod) (*PoolReconciler, *podPatchCountingClient) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + objects := make([]client.Object, 0, len(pods)) + for _, pod := range pods { + objects = append(objects, pod.DeepCopy()) + } + countingClient := &podPatchCountingClient{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build(), + } + return &PoolReconciler{Client: countingClient}, countingClient +} + +func getPodLabelTestPod(t *testing.T, ctx context.Context, c client.Client, pod *corev1.Pod) *corev1.Pod { + t.Helper() + updated := &corev1.Pod{} + require.NoError(t, c.Get(ctx, types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, updated)) + return updated +} + +type podPatchCountingClient struct { + client.Client + mu sync.Mutex + patchCalls int + onPodPatch func() + podPatchErr error +} + +func (c *podPatchCountingClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + err := c.podPatchErr + if err == nil { + err = c.Client.Patch(ctx, obj, patch, opts...) + } + if _, ok := obj.(*corev1.Pod); ok { + c.mu.Lock() + c.patchCalls++ + onPodPatch := c.onPodPatch + c.mu.Unlock() + if err == nil && onPodPatch != nil { + onPodPatch() + } + } + return err +} + +type blockingPodPatchClient struct { + client.Client + started chan struct{} + release chan struct{} +} + +func (c *blockingPodPatchClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error { + if _, ok := obj.(*corev1.Pod); ok { + c.started <- struct{}{} + <-c.release + } + return c.Client.Patch(ctx, obj, patch, opts...) +} + +func (c *podPatchCountingClient) podPatchCalls() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.patchCalls +} + +type labelSyncOrder struct { + mu sync.Mutex + items []string +} + +func (o *labelSyncOrder) record(item string) { + o.mu.Lock() + defer o.mu.Unlock() + o.items = append(o.items, item) +} + +func (o *labelSyncOrder) events() []string { + o.mu.Lock() + defer o.mu.Unlock() + return append([]string(nil), o.items...) +}