Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions kubernetes/internal/controller/events.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@ const (
EventReasonPoolAssigned = "PoolAssigned"
EventReasonFailedPoolAssign = "FailedPoolAssign"

EventReasonFailedSchedule = "FailedSchedule"
EventReasonFailedScale = "FailedScale"
EventReasonFailedUpdate = "FailedUpdate"

// Pod release — recorded on BatchSandbox
EventReasonPodReleased = "PodReleased"
EventReasonFailedRelease = "FailedRelease"
Expand Down
61 changes: 41 additions & 20 deletions kubernetes/internal/controller/pool_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,9 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
// 1. Get latest Pool CR
latestPool := &sandboxv1alpha1.Pool{}
if err := r.Get(ctx, client.ObjectKeyFromObject(pool), latestPool); err != nil {
if errors.IsNotFound(err) {
return nil
}
Comment on lines +196 to +198

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Clean up state when the inner Pool fetch is not found

When a Pool disappears between the initial Reconcile fetch and this retry-closure fetch, returning success bypasses the allocation and scale-expectation cleanup performed by the outer not-found path. Both stores are keyed only by namespace/name, so if a Pool with the same name is recreated before the queued deletion request runs, that request fetches the new object and the old state is never cleared; stale creation expectations can then block the replacement Pool's scaling until timeout, while stale allocations are exposed to its first scheduling pass. Perform the same cleanup here before returning.

Useful? React with 👍 / 👎.

return err
}

Expand All @@ -203,10 +206,11 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
}

// 3. Schedule sandbox (compute + persist + sync)
schedResult, err := r.scheduleSandbox(ctx, latestPool, batchSandboxes, schedulePods)
if err != nil {
return err
schedResult, scheduleSbxErr := r.scheduleSandbox(ctx, latestPool, batchSandboxes, schedulePods)
if scheduleSbxErr != nil {
r.Recorder.Eventf(latestPool, corev1.EventTypeWarning, EventReasonFailedSchedule, "Pool schedule error %v", scheduleSbxErr)
}

// Requeue if there are pending sandboxes waiting for scheduling
if schedResult.SupplyCnt > 0 {
result = ctrl.Result{RequeueAfter: defaultRetryTime}
Expand All @@ -225,9 +229,9 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
}

// 4. Handle pool upgrade
updateResult, err := r.updatePool(ctx, latestPool, schedulePods, schedResult.IdlePods)
if err != nil {
return err
updateResult, updatePoolErr := r.updatePool(ctx, latestPool, schedulePods, schedResult.IdlePods)
if updatePoolErr != nil {
r.Recorder.Eventf(pool, corev1.EventTypeWarning, EventReasonFailedUpdate, "Pool update error %v", updatePoolErr)
}

// 5. Handle pool scale
Expand All @@ -242,20 +246,20 @@ func (r *PoolReconciler) reconcilePool(ctx context.Context, pool *sandboxv1alpha
supplyCnt: schedResult.SupplyCnt + updateResult.SupplyUpdateRevision,
}

if err := r.scalePool(ctx, latestPool, args); err != nil {
return err
scalePoolErr := r.scalePool(ctx, latestPool, args)
if scalePoolErr != nil {
r.Recorder.Eventf(pool, corev1.EventTypeWarning, EventReasonFailedScale, "Pool scale error %v", scalePoolErr)
}

// 6. Update pool status
if err := r.updatePoolStatus(ctx, updateResult.UpdateRevision, latestPool, pods, schedulePods, schedResult.LatestAllocation); err != nil {
return err
}
updatePoolStatusErr := r.updatePoolStatus(ctx, updateResult.UpdateRevision, latestPool, pods, schedulePods, schedResult.LatestAllocation)

if evictionErr != nil {
return evictionErr
for _, err := range []error{evictionErr, scheduleSbxErr, updatePoolErr, scalePoolErr, updatePoolStatusErr} {
if errors.IsConflict(err) {
return err
}
}

return nil
return gerrors.Join(evictionErr, scheduleSbxErr, updatePoolErr, scalePoolErr, updatePoolStatusErr)
})

return result, err
Expand Down Expand Up @@ -771,15 +775,15 @@ func (r *PoolReconciler) scheduleSandbox(ctx context.Context, pool *sandboxv1alp
allocAction, err := r.Allocator.Schedule(ctx, spec)
if err != nil {
r.Recorder.Eventf(pool, corev1.EventTypeWarning, EventReasonAllocationFailed, "Failed to schedule sandboxes: %v", err)
return nil, err
return r.bestEffortScheduleResult(ctx, pool, pods), err
}
log.Info("Allocate action", "pool", pool.Name, "toAllocate", allocAction.ToAllocate, "toRelease", allocAction.ToRelease)

// 2. Execute scheduling actions.
// 2.1 Execute ToAllocate / update in-memory store.
err = r.doAllocate(ctx, pool, batchSandboxes, pods, allocAction.ToAllocate)
if err != nil {
return nil, err
return r.bestEffortScheduleResult(ctx, pool, pods), err
}

// Emit allocation events.
Expand All @@ -802,13 +806,13 @@ func (r *PoolReconciler) scheduleSandbox(ctx context.Context, pool *sandboxv1alp
// 2.2 Execute ToRelease / release in-memory store.
toDeletePods, err := r.doRelease(ctx, pool, batchSandboxes, pods, allocAction.ToRelease)
if err != nil {
return nil, err
return r.bestEffortScheduleResult(ctx, pool, pods), err

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

前面的allocation/release 如果存在部分成功/失败的情况,这里返回的状态会不会导致后面的scale出现问题?

}

// 3. Return schedule result
latestAllocation, err := r.Allocator.GetPoolAllocation(ctx, pool)
if err != nil {
return nil, err
return r.bestEffortScheduleResult(ctx, pool, pods), err
}
idlePods := make([]string, 0)
for _, pod := range pods {
Expand All @@ -827,10 +831,27 @@ func (r *PoolReconciler) scheduleSandbox(ctx context.Context, pool *sandboxv1alp
return result, nil
}

// bestEffortScheduleResult builds a fallback result from the last persisted
// allocation so update/scale/status can proceed when scheduling fails.
func (r *PoolReconciler) bestEffortScheduleResult(ctx context.Context, pool *sandboxv1alpha1.Pool, pods []*corev1.Pod) *ScheduleResult {
allocation, err := r.Allocator.GetPoolAllocation(ctx, pool)
if err != nil {
logf.FromContext(ctx).Error(err, "Failed to read pool allocation for best-effort schedule result", "pool", pool.Name)
allocation = map[string]string{}

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

空map会不会导致后面scale有问题?

}
idlePods := make([]string, 0)
for _, pod := range pods {
if _, ok := allocation[pod.Name]; !ok {
idlePods = append(idlePods, pod.Name)
}
}
return &ScheduleResult{LatestAllocation: allocation, IdlePods: idlePods}
}

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 {
return nil, err
return &UpdateResult{}, err
}
strategy := NewPoolUpdateStrategy(pool)
result := strategy.Compute(ctx, updateRevision, pods, idlePods)
Expand Down
108 changes: 108 additions & 0 deletions kubernetes/internal/controller/pool_schedule_failure_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
// 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"
"testing"

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"

sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1"
"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller/algorithm"
)

type failingScheduleAllocator struct {
stubAllocator
scheduleErr error
}

func (a *failingScheduleAllocator) Schedule(_ context.Context, _ *AllocSpec) (*algorithm.AllocAction, error) {
return nil, a.scheduleErr
}

func newScheduleFailureTestReconciler(alloc Allocator, objs ...runtime.Object) *PoolReconciler {
scheme := runtime.NewScheme()
_ = corev1.AddToScheme(scheme)
_ = sandboxv1alpha1.AddToScheme(scheme)
return &PoolReconciler{
Client: fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objs...).Build(),
Scheme: scheme,
Recorder: record.NewFakeRecorder(10),
Allocator: alloc,
}
}

func TestScheduleSandboxReturnsBestEffortResultOnFailure(t *testing.T) {
ctx := context.Background()
pool := &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool-a", Namespace: "default"}}
pods := []*corev1.Pod{
{ObjectMeta: metav1.ObjectMeta{Name: "pod-allocated", Namespace: "default"}},
{ObjectMeta: metav1.ObjectMeta{Name: "pod-idle", Namespace: "default"}},
}
alloc := &failingScheduleAllocator{
stubAllocator: stubAllocator{podAllocation: map[string]string{"pod-allocated": "sandbox-a"}},
scheduleErr: errors.New("schedule exploded"),
}
r := newScheduleFailureTestReconciler(alloc)

result, err := r.scheduleSandbox(ctx, pool, nil, pods)
if err == nil {
t.Fatal("expected schedule error, got nil")
}
if result == nil {
t.Fatal("expected best-effort result, got nil")
}
if got := result.LatestAllocation["pod-allocated"]; got != "sandbox-a" {
t.Fatalf("LatestAllocation[pod-allocated] = %q, want %q", got, "sandbox-a")
}
if len(result.IdlePods) != 1 || result.IdlePods[0] != "pod-idle" {
t.Fatalf("IdlePods = %v, want [pod-idle]", result.IdlePods)
}
if result.SupplyCnt != 0 || len(result.ToDelete) != 0 {
t.Fatalf("expected zero SupplyCnt/ToDelete on failure, got %+v", result)
}
}

func TestBestEffortScheduleResultFallsBackToEmptyAllocation(t *testing.T) {
ctx := context.Background()
pool := &sandboxv1alpha1.Pool{ObjectMeta: metav1.ObjectMeta{Name: "pool-a", Namespace: "default"}}
pods := []*corev1.Pod{{ObjectMeta: metav1.ObjectMeta{Name: "pod-idle", Namespace: "default"}}}
r := newScheduleFailureTestReconciler(&failingReadAllocator{})

result := r.bestEffortScheduleResult(ctx, pool, pods)
if result == nil {
t.Fatal("expected non-nil result")
}
if len(result.LatestAllocation) != 0 {
t.Fatalf("LatestAllocation = %v, want empty", result.LatestAllocation)
}
if len(result.IdlePods) != 1 || result.IdlePods[0] != "pod-idle" {
t.Fatalf("IdlePods = %v, want [pod-idle]", result.IdlePods)
}
}

type failingReadAllocator struct {
stubAllocator
}

func (a *failingReadAllocator) GetPoolAllocation(_ context.Context, _ *sandboxv1alpha1.Pool) (map[string]string, error) {
return nil, errors.New("store unavailable")
}
Loading