diff --git a/charts/Chart.yaml b/charts/Chart.yaml index 9614a650..4e004057 100644 --- a/charts/Chart.yaml +++ b/charts/Chart.yaml @@ -1,8 +1,8 @@ apiVersion: v1 name: kuperator description: Helm chart for KusionStack Kuperator -version: 0.7.3 -appVersion: 0.7.3 +version: 0.7.4 +appVersion: 0.7.4 home: https://KusionStack.io sources: - https://github.com/KusionStack/kuperator diff --git a/charts/values.yaml b/charts/values.yaml index 95c7d02a..2400c7fc 100644 --- a/charts/values.yaml +++ b/charts/values.yaml @@ -16,7 +16,7 @@ sharding: controlPlane: kusionstack-kuperator image: - tag: v0.7.3 + tag: v0.7.4 repo: kusionstack/kuperator pullPolicy: IfNotPresent diff --git a/pkg/controllers/podtransitionrule/podtransitionrule_controller.go b/pkg/controllers/podtransitionrule/podtransitionrule_controller.go index eabae42f..72b5b70c 100644 --- a/pkg/controllers/podtransitionrule/podtransitionrule_controller.go +++ b/pkg/controllers/podtransitionrule/podtransitionrule_controller.go @@ -98,6 +98,7 @@ type PodTransitionRuleReconciler struct { // +kubebuilder:rbac:groups=apps.kusionstack.io,resources=podtransitionrules,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=apps.kusionstack.io,resources=podtransitionrules/status,verbs=get;update;patch // +kubebuilder:rbac:groups=apps.kusionstack.io,resources=podtransitionrules/finalizers,verbs=update +// +kubebuilder:rbac:groups=apps.kusionstack.io,resources=collasets,verbs=get;list;watch // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;update;patch // +kubebuilder:rbac:groups=core,resources=events,verbs=create;update;patch diff --git a/pkg/controllers/podtransitionrule/processor/rules/available.go b/pkg/controllers/podtransitionrule/processor/rules/available.go index bedf7d14..cc9ca71a 100644 --- a/pkg/controllers/podtransitionrule/processor/rules/available.go +++ b/pkg/controllers/podtransitionrule/processor/rules/available.go @@ -17,10 +17,14 @@ limitations under the License. package rules import ( + "context" "fmt" "time" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/util/sets" appsv1alpha1 "kusionstack.io/kube-api/apps/v1alpha1" @@ -66,8 +70,20 @@ func (r *AvailableRuler) Filter(podTransitionRule *appsv1alpha1.PodTransitionRul } minAvailableQuota = quota } - // TODO: UncreatedReplicas - // allowUnavailable -= uncreatedReplicas + // Subtract uncreated owner replicas so that the quota is not over-granted + // when some pods are temporarily missing. With PersistentSequence naming, + // a rebuild upgrade deletes the old pod before the new one (same name) + // can be created, leaving a window where targets shrink. Reading the + // owner's desired replicas and subtracting the deficit keeps the watermark + // on the desired fleet size instead of the fluctuating live pod count. + uncreatedReplicas, err := r.considerOwnerReplicas(targets) + if err != nil { + return rejectAllWithErr(subjects, pass, rejects, "[%s] fail to count owner uncreated replicas: %v", r.Name, err) + } + allowUnavailable -= uncreatedReplicas + if allowUnavailable < 0 { + allowUnavailable = 0 + } allAvailableSize := 0 var minTimeLeft *int64 // filter unavailable pods @@ -149,6 +165,71 @@ func processUnavailableFunc(pod *corev1.Pod) (bool, *int64) { return isUnavailable, minInterval } +// considerOwnerReplicas counts the deficit between each target pod owner's +// desired replicas and the number of pods currently listed for that owner. +// The returned value is used to shrink the max-unavailable quota so that a +// temporarily shrunken target list (e.g. during a PersistentSequence rebuild +// where the old pod is gone and the new one is not yet created) does not +// inflate the quota and let through more changes than the policy allows. +func (r *AvailableRuler) considerOwnerReplicas(targets map[string]*corev1.Pod) (int, error) { + type ownerStat struct { + desired int + current int + } + ownerStats := map[types.UID]*ownerStat{} + + collasetAPIVersion := appsv1alpha1.SchemeGroupVersion.String() + for _, pod := range targets { + ownerRef := metav1.GetControllerOf(pod) + // Only treat CollaSet owners from our own API group; other groups could + // reuse the same Kind name and would otherwise trigger spurious Gets. + if ownerRef == nil || ownerRef.Kind != "CollaSet" || ownerRef.APIVersion != collasetAPIVersion { + continue + } + + stat, exist := ownerStats[ownerRef.UID] + if !exist { + cls := &appsv1alpha1.CollaSet{} + if err := r.Client.Get(context.TODO(), types.NamespacedName{Namespace: pod.Namespace, Name: ownerRef.Name}, cls); err != nil { + if errors.IsNotFound(err) { + // Owner gone; cache a zero-desired stat so siblings of the + // same owner don't repeat this Get. Zero desired yields a + // non-positive deficit, leaving the quota unchanged. + stat = &ownerStat{desired: 0} + ownerStats[ownerRef.UID] = stat + stat.current++ + continue + } + return 0, fmt.Errorf("fail to get controller CollaSet %s/%s: %w", pod.Namespace, ownerRef.Name, err) + } + // UID mismatch means the ref reused a recycled name; cache a zero + // stat for the same reason as above. + if cls.UID != ownerRef.UID { + stat = &ownerStat{desired: 0} + ownerStats[ownerRef.UID] = stat + stat.current++ + continue + } + desired := 1 + if cls.Spec.Replicas != nil { + desired = int(*cls.Spec.Replicas) + } + stat = &ownerStat{desired: desired} + ownerStats[ownerRef.UID] = stat + } + stat.current++ + } + + uncreated := 0 + for _, stat := range ownerStats { + deficit := stat.desired - stat.current + if deficit > 0 { + uncreated += deficit + } + } + return uncreated, nil +} + func min(a, b *int64) *int64 { if a == nil { return b diff --git a/pkg/controllers/podtransitionrule/processor/rules/available_test.go b/pkg/controllers/podtransitionrule/processor/rules/available_test.go new file mode 100644 index 00000000..9d44eac1 --- /dev/null +++ b/pkg/controllers/podtransitionrule/processor/rules/available_test.go @@ -0,0 +1,304 @@ +/* +Copyright 2026 The KusionStack Authors. + +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 rules + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/apimachinery/pkg/util/sets" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/utils/ptr" + appsv1alpha1 "kusionstack.io/kube-api/apps/v1alpha1" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "kusionstack.io/kuperator/pkg/controllers/podtransitionrule/register" +) + +func newCollaSet(name string, uid types.UID, replicas int32) *appsv1alpha1.CollaSet { + return &appsv1alpha1.CollaSet{ + TypeMeta: metav1.TypeMeta{Kind: "CollaSet", APIVersion: appsv1alpha1.GroupVersion.String()}, + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + UID: uid, + }, + Spec: appsv1alpha1.CollaSetSpec{ + Replicas: ptr.To[int32](replicas), + }, + } +} + +func newOwnedPod(name, ownerName string, ownerUID types.UID) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: appsv1alpha1.GroupVersion.String(), + Kind: "CollaSet", + Name: ownerName, + UID: ownerUID, + Controller: ptr.To(true), + }, + }, + }, + } +} + +func newRuler(objects ...runtime.Object) *AvailableRuler { + scheme := runtime.NewScheme() + _ = clientgoscheme.AddToScheme(scheme) + _ = appsv1alpha1.AddToScheme(scheme) + cl := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objects...).Build() + return &AvailableRuler{ + Client: cl, + } +} + +func TestConsiderOwnerReplicas_NoOwner(t *testing.T) { + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "default"}} + r := newRuler() + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p": pod}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 0 { + t.Fatalf("expected 0 for ownerless pod, got %d", got) + } +} + +func TestConsiderOwnerReplicas_NoDeficit(t *testing.T) { + uid := types.UID("cls-uid") + cls := newCollaSet("cls", uid, 2) + pod1 := newOwnedPod("p1", "cls", uid) + pod2 := newOwnedPod("p2", "cls", uid) + r := newRuler(cls) + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p1": pod1, "p2": pod2}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 0 { + t.Fatalf("expected 0 uncreated when desired==current, got %d", got) + } +} + +func TestConsiderOwnerReplicas_DeficitWhenPodMissing(t *testing.T) { + uid := types.UID("cls-uid") + // Simulate the PersistentSequence rebuild gap: only 1 of the 2 desired pods exists. + cls := newCollaSet("cls", uid, 2) + pod1 := newOwnedPod("p1", "cls", uid) + r := newRuler(cls) + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p1": pod1}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 1 { + t.Fatalf("expected 1 uncreated (desired 2 - current 1), got %d", got) + } +} + +func TestConsiderOwnerReplicas_OwnerNotFound(t *testing.T) { + uid := types.UID("cls-uid") + pod1 := newOwnedPod("p1", "cls", uid) + // No CollaSet object in client; should not error and not contribute. + r := newRuler() + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p1": pod1}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 0 { + t.Fatalf("expected 0 when owner missing, got %d", got) + } +} + +func TestConsiderOwnerReplicas_UIDMismatch(t *testing.T) { + uid := types.UID("cls-uid-old") + cls := newCollaSet("cls", types.UID("cls-uid-new"), 2) + pod1 := newOwnedPod("p1", "cls", uid) + r := newRuler(cls) + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p1": pod1}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 0 { + t.Fatalf("expected 0 on UID mismatch, got %d", got) + } +} + +func TestConsiderOwnerReplicas_MultipleOwners(t *testing.T) { + uidA := types.UID("a") + uidB := types.UID("b") + clsA := newCollaSet("cls-a", uidA, 3) + clsB := newCollaSet("cls-b", uidB, 2) + podA1 := newOwnedPod("a1", "cls-a", uidA) + podB1 := newOwnedPod("b1", "cls-b", uidB) + r := newRuler(clsA, clsB) + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"a1": podA1, "b1": podB1}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + // cls-a: 3-1=2; cls-b: 2-1=1; total 3 + if got != 3 { + t.Fatalf("expected 3 uncreated across owners, got %d", got) + } +} + +func TestConsiderOwnerReplicas_NilReplicasDefaultsOne(t *testing.T) { + uid := types.UID("cls-uid") + cls := newCollaSet("cls", uid, 0) + cls.Spec.Replicas = nil + pod1 := newOwnedPod("p1", "cls", uid) + r := newRuler(cls) + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p1": pod1}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + // desired defaults to 1, current 1, deficit 0 + if got != 0 { + t.Fatalf("expected 0 when nil replicas defaults to 1, got %d", got) + } +} + +func TestConsiderOwnerReplicas_NonCollaSetOwnerIgnored(t *testing.T) { + uid := types.UID("rs-uid") + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "p", + Namespace: "default", + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "apps/v1", + Kind: "ReplicaSet", + Name: "rs", + UID: uid, + Controller: ptr.To(true), + }, + }, + }, + } + r := newRuler() + got, err := r.considerOwnerReplicas(map[string]*corev1.Pod{"p": pod}) + if err != nil { + t.Fatalf("unexpected err: %v", err) + } + if got != 0 { + t.Fatalf("expected 0 for non-CollaSet owner, got %d", got) + } +} + +// TestFilter_PersistentSequenceRebuildGap_ProtectsWatermark reproduces the +// scenario where PersistentSequence rebuild leaves the target list with 1 +// pod while the CollaSet still desires 2. Without the considerOwnerReplicas +// offset the lone remaining pod would be approved (allowing both pods to be +// unavailable simultaneously). With the offset, the deficit (1) shrinks the +// quota to 0 and the remaining subject is rejected. +func TestFilter_PersistentSequenceRebuildGap_ProtectsWatermark(t *testing.T) { + uid := types.UID("cls-uid") + cls := newCollaSet("cls", uid, 2) + // Mark the surviving pod as service-available so it is "available" and + // would otherwise consume a quota slot. + pod1 := newOwnedPod("p1", "cls", uid) + pod1.Labels = map[string]string{appsv1alpha1.PodServiceAvailableLabel: "true"} + pod1.Status.Conditions = []corev1.PodCondition{ + {Type: corev1.ContainersReady, Status: corev1.ConditionTrue}, + {Type: corev1.PodReady, Status: corev1.ConditionTrue}, + } + + r := newRuler(cls) + r.Name = "available" + maxUnavailable := intstr.FromInt(1) + r.MaxUnavailableValue = &maxUnavailable + + // Make processUnavailableFunc treat pods lacking the service-available + // label as unavailable, mirroring the production podopslifecycle hook. + prev := register.UnAvailableFuncList + register.UnAvailableFuncList = []register.UnAvailableFunc{ + func(pod *corev1.Pod) (bool, *int64) { + if pod.Labels == nil || pod.Labels[appsv1alpha1.PodServiceAvailableLabel] != "true" { + return true, nil + } + return false, nil + }, + } + t.Cleanup(func() { register.UnAvailableFuncList = prev }) + + ptrRule := &appsv1alpha1.PodTransitionRule{} + targets := map[string]*corev1.Pod{"p1": pod1} + subjects := sets.NewString("p1") + + result := r.Filter(ptrRule, targets, subjects) + if result == nil { + t.Fatalf("nil filter result") + } + if result.Passed.Has("p1") { + t.Fatalf("p1 must be rejected: deficit should have consumed the quota, got pass=%v rejects=%v", result.Passed.List(), result.Rejected) + } + if _, ok := result.Rejected["p1"]; !ok { + t.Fatalf("p1 must appear in rejects, got rejects=%v", result.Rejected) + } +} + +// TestFilter_FullFleet_NoRegression verifies that when the full fleet exists +// (desired == current), the offset is zero and the legacy behavior is +// preserved: an available subject is approved within the maxUnavailable quota. +func TestFilter_FullFleet_NoRegression(t *testing.T) { + uid := types.UID("cls-uid") + cls := newCollaSet("cls", uid, 2) + pod1 := newOwnedPod("p1", "cls", uid) + pod2 := newOwnedPod("p2", "cls", uid) + // Both pods are service-available. + pod1.Labels = map[string]string{appsv1alpha1.PodServiceAvailableLabel: "true"} + pod2.Labels = map[string]string{appsv1alpha1.PodServiceAvailableLabel: "true"} + for _, p := range []*corev1.Pod{pod1, pod2} { + p.Status.Conditions = []corev1.PodCondition{ + {Type: corev1.ContainersReady, Status: corev1.ConditionTrue}, + {Type: corev1.PodReady, Status: corev1.ConditionTrue}, + } + } + + r := newRuler(cls) + r.Name = "available" + maxUnavailable := intstr.FromInt(1) + r.MaxUnavailableValue = &maxUnavailable + + prev := register.UnAvailableFuncList + register.UnAvailableFuncList = []register.UnAvailableFunc{ + func(pod *corev1.Pod) (bool, *int64) { + if pod.Labels == nil || pod.Labels[appsv1alpha1.PodServiceAvailableLabel] != "true" { + return true, nil + } + return false, nil + }, + } + t.Cleanup(func() { register.UnAvailableFuncList = prev }) + + ptrRule := &appsv1alpha1.PodTransitionRule{} + targets := map[string]*corev1.Pod{"p1": pod1, "p2": pod2} + // Request the transition for p1 (it is available and the quota allows 1). + subjects := sets.NewString("p1") + + result := r.Filter(ptrRule, targets, subjects) + if !result.Passed.Has("p1") { + t.Fatalf("p1 should be approved when fleet is full and quota allows, got rejects=%v", result.Rejected) + } +} diff --git a/test/e2e/apps/collaset.go b/test/e2e/apps/collaset.go index fe8b8f72..14fd724b 100644 --- a/test/e2e/apps/collaset.go +++ b/test/e2e/apps/collaset.go @@ -32,6 +32,7 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/apimachinery/pkg/util/rand" "k8s.io/apimachinery/pkg/util/sets" clientset "k8s.io/client-go/kubernetes" @@ -1642,6 +1643,88 @@ var _ = SIGDescribe("CollaSet", func() { } }) + framework.ConformanceIt("PersistentSequence rebuild upgrade respects PodTransitionRule maxUnavailable", func() { + // Reproduces the gap where PersistentSequence naming deletes the old pod + // before the new one (same name) can be created, leaving the + // PodTransitionRule target list temporarily shrunken. Without the + // owner-replicas offset this would let both pods be unavailable at + // once. The test forces a rebuild (non-image env change) and asserts + // that availableReplicas never drops below 1 during the rolling. + cls := tester.NewCollaSet("collaset-"+randStr, 2, appsv1alpha1.UpdateStrategy{ + PodUpdatePolicy: appsv1alpha1.CollaSetInPlaceIfPossiblePodUpdateStrategyType, + }) + cls.Spec.NamingStrategy = &appsv1alpha1.NamingStrategy{ + PodNamingSuffixPolicy: appsv1alpha1.PodNamingSuffixPolicyPersistentSequence, + } + Expect(tester.CreateCollaSet(cls)).NotTo(HaveOccurred()) + + By("Wait for status replicas satisfied") + Eventually(func() error { return tester.ExpectedStatusReplicas(cls, 2, 2, 2, 2, 2) }, 30*time.Second, 3*time.Second).ShouldNot(HaveOccurred()) + + By("Create PodTransitionRule with maxUnavailable=1 selecting the CollaSet pods") + maxUnavailable := intstr.FromInt(1) + ptr := &appsv1alpha1.PodTransitionRule{ + ObjectMeta: metav1.ObjectMeta{ + Name: "ptr-" + randStr, + Namespace: ns, + }, + Spec: appsv1alpha1.PodTransitionRuleSpec{ + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"owner": cls.Name}}, + Rules: []appsv1alpha1.TransitionRule{ + { + Name: "serviceAvailable", + TransitionRuleDefinition: appsv1alpha1.TransitionRuleDefinition{ + AvailablePolicy: &appsv1alpha1.AvailableRule{ + MaxUnavailableValue: &maxUnavailable, + }, + }, + }, + }, + }, + } + Expect(client.Create(context.TODO(), ptr)).NotTo(HaveOccurred()) + + By("Wait for PodTransitionRule to observe both pods as targets") + Eventually(func() int { + got := &appsv1alpha1.PodTransitionRule{} + if err := client.Get(context.TODO(), types.NamespacedName{Namespace: ptr.Namespace, Name: ptr.Name}, got); err != nil { + return 0 + } + return len(got.Status.Targets) + }, 30*time.Second, 3*time.Second).Should(Equal(2)) + + By("Trigger rebuild upgrade by changing env (not in-place supported)") + Expect(tester.UpdateCollaSet(cls, func(cls *appsv1alpha1.CollaSet) { + cls.Spec.Template.Spec.Containers[0].Env = []v1.EnvVar{ + {Name: "test", Value: "bar"}, + } + })).NotTo(HaveOccurred()) + + By("Poll status, assert availableReplicas never drops below 1, and stop once upgrade completes") + timeout := 120 * time.Second + start := time.Now() + minAvailable := int32(2) + Eventually(func() bool { + Expect(tester.GetCollaSet(cls)).NotTo(HaveOccurred()) + avail := cls.Status.AvailableReplicas + if avail < minAvailable { + minAvailable = avail + } + Expect(avail).To(BeNumerically(">=", int32(1)), fmt.Sprintf("availableReplicas dropped to %d during rolling (elapsed=%v)", avail, time.Since(start))) + return tester.ExpectedStatusReplicas(cls, 2, 2, 2, 2, 2) == nil + }, timeout, time.Second).Should(BeTrue()) + + By(fmt.Sprintf("observed min available replicas during rolling: %d (elapsed=%v)", minAvailable, time.Since(start))) + Expect(minAvailable).To(BeNumerically(">=", int32(1))) + + By("Delete PodTransitionRule so the framework AfterEach can clean up CollaSet cleanly") + Expect(client.Delete(context.TODO(), ptr)).NotTo(HaveOccurred()) + Eventually(func() bool { + err := client.Get(context.TODO(), types.NamespacedName{Namespace: ptr.Namespace, Name: ptr.Name}, &appsv1alpha1.PodTransitionRule{}) + return errors.IsNotFound(err) + }, 120*time.Second, 3*time.Second).Should(BeTrue()) + }) + AfterEach(func() { afterEach(tester, ns) })