wehub-resource-sync
e0e362d700
SDK Tests / changes (push) Successful in 2m29s
Real E2E Tests / changes (push) Successful in 2m29s
Deploy Docs Pages / build (push) Has been cancelled
Deploy Docs Pages / deploy (push) Has been cancelled
Real E2E Tests / JavaScript E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Python E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Java E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / C# E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Go E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Real E2E CI (push) Has been cancelled
SDK Tests / SDK CI (push) Has been cancelled
SDK Tests / CLI Tests (push) Has been cancelled
SDK Tests / Python SDK Quality (code-interpreter) (push) Has been cancelled
SDK Tests / Python SDK Quality (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (sandbox) (push) Has been cancelled
SDK Tests / CLI Quality (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Go SDK Quality And Tests (push) Has been cancelled
1823 lines
63 KiB
Go
1823 lines
63 KiB
Go
// 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"
|
|
"encoding/json"
|
|
stderrors "errors"
|
|
"time"
|
|
|
|
v1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/api/errors"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/fields"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/intstr"
|
|
"k8s.io/apimachinery/pkg/util/rand"
|
|
"k8s.io/client-go/util/retry"
|
|
"k8s.io/utils/ptr"
|
|
kclient "sigs.k8s.io/controller-runtime/pkg/client"
|
|
|
|
"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/controller/recycle"
|
|
"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/utils/fieldindex"
|
|
. "github.com/onsi/ginkgo/v2"
|
|
. "github.com/onsi/gomega"
|
|
|
|
sandboxv1alpha1 "github.com/alibaba/OpenSandbox/sandbox-k8s/apis/sandbox/v1alpha1"
|
|
)
|
|
|
|
var _ = Describe("Pool scale", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When reconciling a resource", func() {
|
|
const resourceName = "pool-scale-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
BeforeEach(func() {
|
|
By("creating the custom resource for the Kind Pool")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
} else {
|
|
By("The specific resource instance Pool already deleted")
|
|
return
|
|
}
|
|
}
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
})
|
|
It("should successfully update pool status", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Eventually(func(g Gomega) {
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
|
|
return
|
|
}
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
It("should successfully scale out pool buffer size", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pool.Spec.CapacitySpec.BufferMin = 2
|
|
pool.Spec.CapacitySpec.BufferMax = 2
|
|
pool.Spec.ScaleStrategy = &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
}
|
|
Expect(k8sClient.Update(ctx, pool)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
|
|
return
|
|
}
|
|
cnt := int32(2)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
It("should successfully scale out buffer limit by pool max", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pool.Spec.CapacitySpec.PoolMax = 2
|
|
pool.Spec.CapacitySpec.BufferMin = 3
|
|
pool.Spec.CapacitySpec.BufferMax = 3
|
|
pool.Spec.ScaleStrategy = &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
}
|
|
Expect(k8sClient.Update(ctx, pool)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
|
|
return
|
|
}
|
|
cnt := int32(2)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
It("should successfully scale in pool buffer size", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pool.Spec.CapacitySpec.BufferMin = 0
|
|
pool.Spec.CapacitySpec.BufferMax = 0
|
|
Expect(k8sClient.Update(ctx, pool)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
|
|
return
|
|
}
|
|
cnt := int32(0)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
It("should successfully scale in buffer limit by pool min", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pool.Spec.CapacitySpec.PoolMax = 1
|
|
pool.Spec.CapacitySpec.PoolMin = 1
|
|
pool.Spec.CapacitySpec.BufferMin = 0
|
|
pool.Spec.CapacitySpec.BufferMax = 0
|
|
Expect(k8sClient.Update(ctx, pool)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, pool); err != nil {
|
|
return
|
|
}
|
|
cnt := int32(1)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
var _ = Describe("Pool update", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When reconciling a resource", func() {
|
|
const resourceName = "pool-update-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
|
|
BeforeEach(func() {
|
|
By("creating the custom resource for the Kind Pool")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
// Mock pod running and ready
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
} else {
|
|
By("The specific resource instance Pool already deleted")
|
|
return
|
|
}
|
|
}
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
})
|
|
It("should successfully update pool revision", func() {
|
|
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())
|
|
cnt := int32(1)
|
|
g.Expect(pool.Status.Revision).NotTo(Equal(oldRevision))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
It("should successfully update pool with allocated pod", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
sbxNamespaceName := types.NamespacedName{
|
|
Name: "sandbox-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
sandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: sbxNamespaceName.Name,
|
|
Namespace: sbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, sandbox)).To(Succeed())
|
|
// wait allocation
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, sbxNamespaceName, sandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(sandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
}, timeout, interval).Should(Succeed())
|
|
Expect(k8sClient.Get(ctx, sbxNamespaceName, sandbox)).To(Succeed())
|
|
sbxAlloc, err := getSandboxAllocation(sandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(sbxAlloc.Pods)).To(Equal(1))
|
|
// check pool allocation
|
|
err = k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(allocation.PodAllocation)).To(Equal(1))
|
|
Expect(allocation.PodAllocation[sbxAlloc.Pods[0]]).To(Equal(sandbox.Name))
|
|
// update pool
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
oldRevision := pool.Status.Revision
|
|
pool.Spec.Template.Labels = map[string]string{
|
|
"test.pool.update": "v1",
|
|
}
|
|
pool.Spec.ScaleStrategy = &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
}
|
|
Expect(k8sClient.Update(ctx, pool)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
cnt := int32(2)
|
|
g.Expect(pool.Status.Revision).NotTo(Equal(oldRevision))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
if pod.Name == sbxAlloc.Pods[0] {
|
|
g.Expect(pod.DeletionTimestamp).To(BeNil())
|
|
g.Expect(pod.Labels[LabelPoolRevision]).To(Equal(oldRevision))
|
|
continue
|
|
}
|
|
if pod.DeletionTimestamp != nil {
|
|
continue
|
|
}
|
|
g.Expect(pod.Labels[LabelPoolRevision]).NotTo(Equal(oldRevision))
|
|
}
|
|
}, timeout, interval).Should(Succeed())
|
|
Expect(k8sClient.Delete(ctx, sandbox)).To(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
var _ = Describe("Pool allocate", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When reconciling a resource", func() {
|
|
const resourceName = "pool-allocate-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
|
|
BeforeEach(func() {
|
|
By("creating the custom resource for the Kind Pool")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
// Mock pod running and ready
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
} else {
|
|
By("The specific resource instance Pool already deleted")
|
|
return
|
|
}
|
|
}
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
})
|
|
It("should successfully allocate pool pod to batch sandbox and release", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
// wait allocation
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
}, timeout, interval).Should(Succeed())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
sbxAlloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(sbxAlloc.Pods)).To(Equal(1))
|
|
// check pool allocation
|
|
err = k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(allocation.PodAllocation)).To(Equal(1))
|
|
Expect(allocation.PodAllocation[sbxAlloc.Pods[0]]).To(Equal(batchSandbox.Name))
|
|
// release
|
|
release := AllocationRelease{
|
|
Pods: sbxAlloc.Pods,
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
err = k8sClient.Update(ctx, batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
// wait release
|
|
Eventually(func(g Gomega) {
|
|
err = k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
allocation, err = getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(len(allocation.PodAllocation)).To(Equal(0))
|
|
}, timeout, interval).Should(Succeed())
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should GC allocation when a pooled BatchSandbox detaches from its pool", func() {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-solidify-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
var allocatedPod string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).To(HaveLen(1))
|
|
allocatedPod = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
Expect(allocatedPod).NotTo(BeEmpty())
|
|
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
latest := &sandboxv1alpha1.BatchSandbox{}
|
|
if err := k8sClient.Get(ctx, bsbxNamespaceName, latest); err != nil {
|
|
return err
|
|
}
|
|
latest.Spec.Template = pool.Spec.Template.DeepCopy()
|
|
latest.Spec.PoolRef = ""
|
|
return k8sClient.Update(ctx, latest)
|
|
})).To(Succeed())
|
|
|
|
maxUnavailable := intstr.FromString("100%")
|
|
Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
|
latestPool := &sandboxv1alpha1.Pool{}
|
|
if err := k8sClient.Get(ctx, typeNamespacedName, latestPool); err != nil {
|
|
return err
|
|
}
|
|
latestPool.Spec.ScaleStrategy = &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: &maxUnavailable,
|
|
}
|
|
return k8sClient.Update(ctx, latestPool)
|
|
})).To(Succeed())
|
|
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
allocation, err := getPoolAllocation(pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(allocation.PodAllocation).NotTo(HaveKey(allocatedPod))
|
|
|
|
pod := &v1.Pod{}
|
|
err = k8sClient.Get(ctx, types.NamespacedName{
|
|
Name: allocatedPod,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}, pod)
|
|
if err == nil {
|
|
g.Expect(pod.DeletionTimestamp).NotTo(BeNil())
|
|
} else {
|
|
g.Expect(errors.IsNotFound(err)).To(BeTrue())
|
|
}
|
|
}, 5*time.Second, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
func getSandboxAllocation(obj kclient.Object) (*SandboxAllocation, error) {
|
|
allocation := &SandboxAllocation{}
|
|
anno := obj.GetAnnotations()
|
|
if anno == nil {
|
|
return allocation, nil
|
|
}
|
|
str, ok := anno[AnnoAllocStatusKey]
|
|
if !ok {
|
|
return allocation, nil
|
|
}
|
|
err := json.Unmarshal([]byte(str), allocation)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return allocation, nil
|
|
}
|
|
|
|
func getPoolAllocation(pool *sandboxv1alpha1.Pool) (*PoolAllocation, error) {
|
|
store := NewInMemoryAllocationStore()
|
|
if err := store.Recover(ctx, k8sClient); err != nil {
|
|
return nil, err
|
|
}
|
|
return store.GetAllocation(ctx, pool)
|
|
}
|
|
|
|
var _ = Describe("Pool deletion and recreation", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When deleting and recreating a Pool with same name", func() {
|
|
const resourceName = "pool-recreate-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
|
|
BeforeEach(func() {
|
|
By("creating the custom resource for the Kind Pool")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
}
|
|
} else {
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
}
|
|
})
|
|
|
|
It("should allow recreating a Pool with the same name after deletion", func() {
|
|
By("deleting the existing Pool")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
Expect(k8sClient.Delete(ctx, pool)).To(Succeed())
|
|
|
|
By("waiting for the Pool to be fully deleted")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(errors.IsNotFound(err)).To(BeTrue(), "Pool should be deleted")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("recreating a Pool with the same name")
|
|
newPool := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, newPool)).To(Succeed())
|
|
|
|
By("verifying the new Pool is successfully reconciled and creates expected pods")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt), "new Pool should have correct total pod count")
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
|
|
It("should clear allocation when recreating a Pool with the same name", func() {
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
// Mock pod running and ready
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox to allocate pods from the pool")
|
|
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-recreate-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation to complete")
|
|
var allocatedPod string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPod = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("verifying pool allocation exists")
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(allocation.PodAllocation)).To(Equal(1))
|
|
Expect(allocation.PodAllocation[allocatedPod]).To(Equal(batchSandbox.Name))
|
|
|
|
By("deleting the BatchSandbox")
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
Eventually(func(g Gomega) {
|
|
err := k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)
|
|
g.Expect(errors.IsNotFound(err)).To(BeTrue())
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("deleting the Pool")
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
Expect(k8sClient.Delete(ctx, pool)).To(Succeed())
|
|
|
|
By("waiting for the Pool to be fully deleted")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(errors.IsNotFound(err)).To(BeTrue(), "Pool should be deleted")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("recreating a Pool with the same name")
|
|
newPool := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, newPool)).To(Succeed())
|
|
|
|
By("verifying the new Pool has empty allocation")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
allocation, err := getPoolAllocation(pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(len(allocation.PodAllocation)).To(Equal(0), "new Pool should have empty allocation, not reuse old allocation")
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
var _ = Describe("Pool scaleStrategy", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When reconciling a resource with scaleStrategy", func() {
|
|
const resourceName = "pool-scale-strategy-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
|
|
BeforeEach(func() {
|
|
By("creating the custom resource for the Kind Pool")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
} else {
|
|
By("The specific resource instance Pool already deleted")
|
|
return
|
|
}
|
|
}
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
})
|
|
|
|
It("should scale with default 25% maxUnavailable when not specified", func() {
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 10,
|
|
BufferMin: 10,
|
|
BufferMax: 10,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
// With default 25% maxUnavailable and target 10 pods,
|
|
// first batch should create at most 3 pods (ceil(10 * 0.25) = 3)
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
// Initial creation should be limited to 3 (or less if pods become ready)
|
|
g.Expect(pool.Status.Total).To(BeNumerically("<=", 3))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
|
|
It("should scale with percentage maxUnavailable", func() {
|
|
maxUnavailable := intstr.FromString("20%")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 10,
|
|
BufferMin: 10,
|
|
BufferMax: 10,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: &maxUnavailable,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
// With 20% maxUnavailable and target 10 pods,
|
|
// first batch should create at most 2 pods
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(pool.Status.Total).To(BeNumerically("<=", 2))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
|
|
It("should scale with integer maxUnavailable", func() {
|
|
maxUnavailable := intstr.FromInt(1)
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 5,
|
|
BufferMin: 5,
|
|
BufferMax: 5,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: &maxUnavailable,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
// With maxUnavailable=1, only 1 pod should be created at a time
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(pool.Status.Total).To(BeNumerically("<=", 1))
|
|
}, timeout, interval).Should(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
var _ = Describe("Pool recycle", func() {
|
|
var (
|
|
timeout = 10 * time.Second
|
|
interval = 1 * time.Second
|
|
)
|
|
Context("When reconciling a resource with RecycleStrategy", func() {
|
|
const resourceName = "pool-recycle-test"
|
|
|
|
ctx := context.Background()
|
|
|
|
typeNamespacedName := types.NamespacedName{
|
|
Name: resourceName,
|
|
Namespace: "default",
|
|
}
|
|
|
|
BeforeEach(func() {
|
|
By("randomizing pool name for isolation")
|
|
typeNamespacedName.Name = resourceName + "-" + rand.String(8)
|
|
})
|
|
|
|
AfterEach(func() {
|
|
resource := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, resource)
|
|
if err != nil {
|
|
if !errors.IsNotFound(err) {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
} else {
|
|
By("The specific resource instance Pool already deleted")
|
|
return
|
|
}
|
|
}
|
|
By("Cleanup the specific resource instance Pool")
|
|
Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
|
|
})
|
|
|
|
It("should create pool with Delete recycle strategy and release pods via deletion", func() {
|
|
By("creating a Pool with explicit Delete RecycleStrategy")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
RecycleStrategy: &sandboxv1alpha1.RecycleStrategy{
|
|
Type: sandboxv1alpha1.RecycleTypeDelete,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox to allocate pods from the pool")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-recycle-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation")
|
|
var allocatedPodName string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPodName = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing pods by updating the BatchSandbox alloc-release annotation")
|
|
release := AllocationRelease{
|
|
Pods: []string{allocatedPodName},
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying the pool allocation is cleared after Delete recycle")
|
|
Eventually(func(g Gomega) {
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(len(allocation.PodAllocation)).To(Equal(0), "Pool allocation should be cleared after Delete recycle")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("verifying the released pod is deleted with Delete recycle strategy")
|
|
Eventually(func(g Gomega) {
|
|
pod := &v1.Pod{}
|
|
err := k8sClient.Get(ctx, types.NamespacedName{Name: allocatedPodName, Namespace: typeNamespacedName.Namespace}, pod)
|
|
if err != nil && errors.IsNotFound(err) {
|
|
return // pod already gone
|
|
}
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(pod.DeletionTimestamp).NotTo(BeNil(), "pod should be terminating with Delete recycle strategy")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should create pool with default (no explicit) RecycleStrategy which behaves as Delete", func() {
|
|
By("creating a Pool without explicit RecycleStrategy")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 2,
|
|
BufferMin: 1,
|
|
BufferMax: 1,
|
|
},
|
|
// No RecycleStrategy set - should default to Delete
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox to allocate pods from the pool")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-default-recycle-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation")
|
|
var allocatedPodName string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPodName = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing pods by updating the BatchSandbox alloc-release annotation")
|
|
release := AllocationRelease{
|
|
Pods: []string{allocatedPodName},
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying the released pod is deleted (default RecycleStrategy = Delete)")
|
|
Eventually(func(g Gomega) {
|
|
pod := &v1.Pod{}
|
|
err := k8sClient.Get(ctx, types.NamespacedName{Name: allocatedPodName, Namespace: typeNamespacedName.Namespace}, pod)
|
|
if err != nil && errors.IsNotFound(err) {
|
|
return // pod already gone
|
|
}
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(pod.DeletionTimestamp).NotTo(BeNil(), "pod should be terminating with default Delete recycle strategy")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should delete multiple pods when BatchSandbox releases with Delete recycle strategy", func() {
|
|
By("creating a Pool with Delete RecycleStrategy and larger capacity")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 4,
|
|
BufferMin: 2,
|
|
BufferMax: 2,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
},
|
|
RecycleStrategy: &sandboxv1alpha1.RecycleStrategy{
|
|
Type: sandboxv1alpha1.RecycleTypeDelete,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox with 2 replicas")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-multi-delete-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(2)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation of 2 pods")
|
|
var allocatedPodNames []string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).To(HaveLen(2))
|
|
allocatedPodNames = alloc.Pods
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing all pods")
|
|
release := AllocationRelease{
|
|
Pods: allocatedPodNames,
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying all released pods are deleted")
|
|
Eventually(func(g Gomega) {
|
|
for _, podName := range allocatedPodNames {
|
|
pod := &v1.Pod{}
|
|
err := k8sClient.Get(ctx, types.NamespacedName{Name: podName, Namespace: typeNamespacedName.Namespace}, pod)
|
|
if err != nil && errors.IsNotFound(err) {
|
|
continue
|
|
}
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(pod.DeletionTimestamp).NotTo(BeNil(), "pod %s should be terminating", podName)
|
|
}
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should create pool with Noop recycle strategy and keep pods available after release", func() {
|
|
By("creating a Pool with Noop RecycleStrategy")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 2,
|
|
PoolMax: 4,
|
|
BufferMin: 2,
|
|
BufferMax: 2,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
},
|
|
RecycleStrategy: &sandboxv1alpha1.RecycleStrategy{
|
|
Type: sandboxv1alpha1.RecycleTypeNoop,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox to allocate pods from the pool")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-noop-recycle-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation")
|
|
var allocatedPodName string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPodName = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing pods by updating the BatchSandbox alloc-release annotation")
|
|
release := AllocationRelease{
|
|
Pods: []string{allocatedPodName},
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying the pool allocation is cleared after Noop recycle")
|
|
Eventually(func(g Gomega) {
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(len(allocation.PodAllocation)).To(Equal(0), "Pool allocation should be cleared after Noop recycle")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("verifying the released pod still exists immediately after Noop recycle (not deleted by recycler)")
|
|
// Unlike Delete recycle which marks pods for deletion, Noop recycle leaves pods intact.
|
|
// The pool's scale-down logic may subsequently delete pods if buffer exceeds BufferMax,
|
|
// but that is a separate concern from the recycle strategy.
|
|
Eventually(func(g Gomega) {
|
|
pod := &v1.Pod{}
|
|
err := k8sClient.Get(ctx, types.NamespacedName{Name: allocatedPodName, Namespace: typeNamespacedName.Namespace}, pod)
|
|
if err != nil && errors.IsNotFound(err) {
|
|
// The pod might have been deleted by pool scale-down, which is acceptable.
|
|
// The key difference from Delete recycle is that Noop recycler does NOT
|
|
// return NeedDelete=true, so the pod is not targeted for deletion by recycle itself.
|
|
return
|
|
}
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
// If the pod still exists, it should not have DeletionTimestamp from recycle.
|
|
// If it was deleted, it was deleted by pool scale-down, not by the recycler.
|
|
g.Expect(pod.DeletionTimestamp).To(BeNil(), "pod should not be marked for deletion by Noop recycle")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("verifying pool available count increases after release")
|
|
Eventually(func(g Gomega) {
|
|
err := k8sClient.Get(ctx, typeNamespacedName, pool)
|
|
g.Expect(err).NotTo(HaveOccurred())
|
|
// Pool.Available should have increased since the released pod is available for reallocation
|
|
g.Expect(pool.Status.Allocated).To(BeNumerically("<", 1), "Pool allocated should be 0 after release")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should create pool with Restart recycle strategy and keep pod alive after release", func() {
|
|
By("creating a Pool with Restart RecycleStrategy")
|
|
// In envtest there is no real kubelet, so the exec command never actually runs.
|
|
// The restart handler will keep retrying until max retries is exceeded and fall
|
|
// back to NeedDelete=true. We therefore verify the same observable end-state as
|
|
// Delete recycle (allocation cleared, pod eventually terminating), while confirming
|
|
// that the Restart strategy path is exercised through the factory.
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 2,
|
|
PoolMax: 4,
|
|
BufferMin: 2,
|
|
BufferMax: 2,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
},
|
|
RecycleStrategy: &sandboxv1alpha1.RecycleStrategy{
|
|
Type: sandboxv1alpha1.RecycleTypeRestart,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
g.Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
cnt := min(pool.Spec.CapacitySpec.PoolMax, pool.Spec.CapacitySpec.BufferMin)
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(cnt))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox to allocate pods from the pool")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-restart-recycle-test-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for allocation")
|
|
var allocatedPodName string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPodName = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing pods by updating the BatchSandbox alloc-release annotation")
|
|
release := AllocationRelease{
|
|
Pods: []string{allocatedPodName},
|
|
}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying the pool allocation is cleared after Restart recycle (retries exceeded → NeedDelete)")
|
|
Eventually(func(g Gomega) {
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(allocation.PodAllocation).To(BeEmpty(), "Pool allocation should be cleared after Restart recycle")
|
|
}, 30*time.Second, interval).Should(Succeed())
|
|
|
|
By("verifying pool Allocated count returns to 0")
|
|
Eventually(func(g Gomega) {
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
g.Expect(pool.Status.Allocated).To(BeNumerically("==", 0), "Pool allocated should be 0 after restart recycle")
|
|
}, 30*time.Second, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
|
|
It("should replenish buffer after Delete recycle frees pods", func() {
|
|
By("creating a Pool with Delete RecycleStrategy and BufferMin=2")
|
|
resource := &sandboxv1alpha1.Pool{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: typeNamespacedName.Name,
|
|
Namespace: typeNamespacedName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.PoolSpec{
|
|
Template: &v1.PodTemplateSpec{
|
|
Spec: v1.PodSpec{
|
|
Containers: []v1.Container{
|
|
{
|
|
Name: "main",
|
|
Image: "example.com",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
CapacitySpec: sandboxv1alpha1.CapacitySpec{
|
|
PoolMin: 0,
|
|
PoolMax: 4,
|
|
BufferMin: 2,
|
|
BufferMax: 2,
|
|
},
|
|
ScaleStrategy: &sandboxv1alpha1.ScaleStrategy{
|
|
MaxUnavailable: ptr.To(intstr.FromString("100%")),
|
|
},
|
|
RecycleStrategy: &sandboxv1alpha1.RecycleStrategy{
|
|
Type: sandboxv1alpha1.RecycleTypeDelete,
|
|
},
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, resource)).To(Succeed())
|
|
|
|
By("waiting for Pool to be ready with 2 buffer pods")
|
|
Eventually(func(g Gomega) {
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
g.Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
g.Expect(pool.Status.ObservedGeneration).To(Equal(pool.Generation))
|
|
g.Expect(pool.Status.Total).To(Equal(int32(2)))
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("setting pool pods to running state")
|
|
pool := &sandboxv1alpha1.Pool{}
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
pods := &v1.PodList{}
|
|
Expect(k8sClient.List(ctx, pods, &kclient.ListOptions{
|
|
Namespace: typeNamespacedName.Namespace,
|
|
FieldSelector: fields.SelectorFromSet(fields.Set{fieldindex.IndexNameForOwnerRefUID: string(pool.UID)}),
|
|
})).To(Succeed())
|
|
for _, pod := range pods.Items {
|
|
pod.Status.Phase = v1.PodRunning
|
|
pod.Status.Conditions = []v1.PodCondition{{Type: v1.PodReady, Status: v1.ConditionTrue}}
|
|
Expect(k8sClient.Status().Update(ctx, &pod)).To(Succeed())
|
|
}
|
|
|
|
By("creating a BatchSandbox with 1 replica")
|
|
bsbxNamespaceName := types.NamespacedName{
|
|
Name: "batch-sandbox-replenish-" + rand.String(8),
|
|
Namespace: typeNamespacedName.Namespace,
|
|
}
|
|
batchSandbox := &sandboxv1alpha1.BatchSandbox{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: bsbxNamespaceName.Name,
|
|
Namespace: bsbxNamespaceName.Namespace,
|
|
},
|
|
Spec: sandboxv1alpha1.BatchSandboxSpec{
|
|
Replicas: ptr.To(int32(1)),
|
|
PoolRef: typeNamespacedName.Name,
|
|
},
|
|
}
|
|
Expect(k8sClient.Create(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("waiting for 1 pod to be allocated")
|
|
var allocatedPodName string
|
|
Eventually(func(g Gomega) {
|
|
g.Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
alloc, err := getSandboxAllocation(batchSandbox)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(alloc.Pods).NotTo(BeEmpty())
|
|
allocatedPodName = alloc.Pods[0]
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("releasing the allocated pod")
|
|
release := AllocationRelease{Pods: []string{allocatedPodName}}
|
|
js, err := json.Marshal(release)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(k8sClient.Get(ctx, bsbxNamespaceName, batchSandbox)).To(Succeed())
|
|
if batchSandbox.Annotations == nil {
|
|
batchSandbox.Annotations = make(map[string]string)
|
|
}
|
|
batchSandbox.Annotations[AnnoAllocReleaseKey] = string(js)
|
|
Expect(k8sClient.Update(ctx, batchSandbox)).To(Succeed())
|
|
|
|
By("verifying pool allocation is cleared")
|
|
Eventually(func(g Gomega) {
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
allocation, err := getPoolAllocation(pool)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
g.Expect(allocation.PodAllocation).To(BeEmpty(), "Pool allocation should be cleared after Delete recycle")
|
|
}, timeout, interval).Should(Succeed())
|
|
|
|
By("verifying pool schedules a new pod to replenish the deleted buffer pod")
|
|
// The deleted pod reduces Total; the reconciler should scale up to restore BufferMin=2.
|
|
// envtest does not run kubelet so new pods stay Pending, but the scale-up expectation
|
|
// being satisfied is enough for Total to reflect the new pod.
|
|
Eventually(func(g Gomega) {
|
|
Expect(k8sClient.Get(ctx, typeNamespacedName, pool)).To(Succeed())
|
|
// After the deleted pod is gone and a replacement is created, Total >= 2.
|
|
g.Expect(pool.Status.Total).To(BeNumerically(">=", int32(2)),
|
|
"pool should replenish buffer after pod deletion")
|
|
}, 30*time.Second, interval).Should(Succeed())
|
|
|
|
Expect(k8sClient.Delete(ctx, batchSandbox)).To(Succeed())
|
|
})
|
|
})
|
|
})
|
|
|
|
var _ = Describe("collectRecycleResults", func() {
|
|
type testCase struct {
|
|
desc string
|
|
results []recycleResult
|
|
wantSucceedMap map[string][]string
|
|
wantToDelete []string
|
|
wantErrContains string
|
|
}
|
|
|
|
cases := []testCase{
|
|
{
|
|
desc: "empty results returns empty maps",
|
|
results: nil,
|
|
wantSucceedMap: map[string][]string{},
|
|
wantToDelete: nil,
|
|
},
|
|
{
|
|
desc: "Succeeded state adds pod to succeedMap",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-1", podName: "pod-1", status: &recycle.Status{State: recycle.StateSucceeded}},
|
|
},
|
|
wantSucceedMap: map[string][]string{"sbx-1": {"pod-1"}},
|
|
wantToDelete: nil,
|
|
},
|
|
{
|
|
desc: "NeedDelete adds pod to toDeletePods and succeedMap",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-1", podName: "pod-1", status: &recycle.Status{State: recycle.StateSucceeded, NeedDelete: true}},
|
|
},
|
|
wantSucceedMap: map[string][]string{"sbx-1": {"pod-1"}},
|
|
wantToDelete: []string{"pod-1"},
|
|
},
|
|
{
|
|
desc: "Recycling state (in-progress) does not add to succeedMap or toDeletePods",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-1", podName: "pod-1", status: &recycle.Status{State: recycle.StateRecycling}},
|
|
},
|
|
wantSucceedMap: map[string][]string{},
|
|
wantToDelete: nil,
|
|
},
|
|
{
|
|
desc: "error result is collected and pod is skipped",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-1", podName: "pod-1", err: stderrors.New("exec failed")},
|
|
},
|
|
wantSucceedMap: map[string][]string{},
|
|
wantToDelete: nil,
|
|
wantErrContains: "exec failed",
|
|
},
|
|
{
|
|
desc: "multiple sandboxes multiple pods",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-a", podName: "pod-1", status: &recycle.Status{State: recycle.StateSucceeded}},
|
|
{sandboxName: "sbx-a", podName: "pod-2", status: &recycle.Status{State: recycle.StateSucceeded, NeedDelete: true}},
|
|
{sandboxName: "sbx-b", podName: "pod-3", status: &recycle.Status{State: recycle.StateSucceeded}},
|
|
{sandboxName: "sbx-b", podName: "pod-4", err: stderrors.New("timeout")},
|
|
},
|
|
wantSucceedMap: map[string][]string{
|
|
"sbx-a": {"pod-1", "pod-2"},
|
|
"sbx-b": {"pod-3"},
|
|
},
|
|
wantToDelete: []string{"pod-2"},
|
|
wantErrContains: "timeout",
|
|
},
|
|
{
|
|
desc: "multiple errors are joined",
|
|
results: []recycleResult{
|
|
{sandboxName: "sbx-1", podName: "pod-1", err: stderrors.New("err-a")},
|
|
{sandboxName: "sbx-1", podName: "pod-2", err: stderrors.New("err-b")},
|
|
},
|
|
wantSucceedMap: map[string][]string{},
|
|
wantToDelete: nil,
|
|
wantErrContains: "err-a",
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
tc := tc
|
|
It(tc.desc, func() {
|
|
succeedMap, toDeletePods, err := collectRecycleResults(context.Background(), tc.results)
|
|
if tc.wantErrContains != "" {
|
|
Expect(err).To(HaveOccurred())
|
|
Expect(err.Error()).To(ContainSubstring(tc.wantErrContains))
|
|
} else {
|
|
Expect(err).NotTo(HaveOccurred())
|
|
}
|
|
Expect(succeedMap).To(Equal(tc.wantSucceedMap))
|
|
Expect(toDeletePods).To(Equal(tc.wantToDelete))
|
|
})
|
|
}
|
|
})
|