From c8e8bd5085d61138270f9660e09f314372bf1eda Mon Sep 17 00:00:00 2001 From: Michael Aspinwall Date: Thu, 26 Feb 2026 06:17:31 +0000 Subject: [PATCH] Add the ability for the statefulset controller to read its own writes --- pkg/controller/statefulset/metrics/metrics.go | 11 ++ .../statefulset/stateful_pod_control.go | 80 ++++++-- .../statefulset/stateful_pod_control_test.go | 180 ++++++++++++++++-- pkg/controller/statefulset/stateful_set.go | 78 +++++++- .../statefulset/stateful_set_control_test.go | 8 +- .../stateful_set_status_updater.go | 21 +- .../stateful_set_status_updater_test.go | 46 ++++- .../statefulset/stateful_set_test.go | 4 +- pkg/features/kube_features.go | 13 ++ .../reference/feature_list.md | 1 + .../reference/versioned_feature_list.yaml | 6 + 11 files changed, 386 insertions(+), 62 deletions(-) diff --git a/pkg/controller/statefulset/metrics/metrics.go b/pkg/controller/statefulset/metrics/metrics.go index ff530a34262..1c4bc167c54 100644 --- a/pkg/controller/statefulset/metrics/metrics.go +++ b/pkg/controller/statefulset/metrics/metrics.go @@ -62,6 +62,17 @@ var ( StabilityLevel: metrics.ALPHA, }, []string{"statefulset_namespace", "statefulset_name", "pod_management_policy"}, ) + + // StatefulSetRequeueSkips tracks the number of StatefulSet syncs skipped due to a stale watch cache. + StatefulSetRequeueSkips = metrics.NewCounterVec( + &metrics.CounterOpts{ + Subsystem: StatefulSetControllerSubsystem, + Name: "stale_sync_skips_total", + Help: "Total number of StatefulSet syncs skipped due to a stale watch cache.", + StabilityLevel: metrics.ALPHA, + }, + []string{"group", "resource"}, + ) ) var registerMetrics sync.Once diff --git a/pkg/controller/statefulset/stateful_pod_control.go b/pkg/controller/statefulset/stateful_pod_control.go index 042c5614a3d..cb52896a817 100644 --- a/pkg/controller/statefulset/stateful_pod_control.go +++ b/pkg/controller/statefulset/stateful_pod_control.go @@ -26,6 +26,7 @@ import ( v1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" errorutils "k8s.io/apimachinery/pkg/util/errors" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientset "k8s.io/client-go/kubernetes" @@ -33,18 +34,19 @@ import ( "k8s.io/client-go/tools/record" "k8s.io/client-go/util/retry" "k8s.io/klog/v2" + consistencyutil "k8s.io/kubernetes/pkg/controller/util/consistency" ) // StatefulPodControlObjectManager abstracts the manipulation of Pods and PVCs. The real controller implements this // with a clientset for writes and listers for reads; for tests we provide stubs. type StatefulPodControlObjectManager interface { - CreatePod(ctx context.Context, pod *v1.Pod) error + CreatePod(ctx context.Context, pod *v1.Pod, ss *apps.StatefulSet) error GetPod(namespace, podName string) (*v1.Pod, error) - UpdatePod(pod *v1.Pod) error + UpdatePod(pod *v1.Pod, ss *apps.StatefulSet) error DeletePod(pod *v1.Pod) error - CreateClaim(claim *v1.PersistentVolumeClaim) error + CreateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error GetClaim(namespace, claimName string) (*v1.PersistentVolumeClaim, error) - UpdateClaim(claim *v1.PersistentVolumeClaim) error + UpdateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error } // StatefulPodControl defines the interface that StatefulSetController uses to create, update, and delete Pods, @@ -63,8 +65,9 @@ func NewStatefulPodControl( podLister corelisters.PodLister, claimLister corelisters.PersistentVolumeClaimLister, recorder record.EventRecorder, + consistencyStore consistencyutil.ConsistencyStore, ) *StatefulPodControl { - return &StatefulPodControl{&realStatefulPodControlObjectManager{client, podLister, claimLister}, recorder} + return &StatefulPodControl{&realStatefulPodControlObjectManager{client, podLister, claimLister, consistencyStore}, recorder} } // NewStatefulPodControlFromManager creates a StatefulPodControl using the given StatefulPodControlObjectManager and recorder. @@ -74,13 +77,23 @@ func NewStatefulPodControlFromManager(om StatefulPodControlObjectManager, record // realStatefulPodControlObjectManager uses a clientset.Interface and listers. type realStatefulPodControlObjectManager struct { - client clientset.Interface - podLister corelisters.PodLister - claimLister corelisters.PersistentVolumeClaimLister + client clientset.Interface + podLister corelisters.PodLister + claimLister corelisters.PersistentVolumeClaimLister + consistencyStore consistencyutil.ConsistencyStore } -func (om *realStatefulPodControlObjectManager) CreatePod(ctx context.Context, pod *v1.Pod) error { - _, err := om.client.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{}) +func (om *realStatefulPodControlObjectManager) CreatePod(ctx context.Context, pod *v1.Pod, ss *apps.StatefulSet) error { + pod, err := om.client.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{}) + if err == nil { + ssNamespacedName := types.NamespacedName{Namespace: ss.Namespace, Name: ss.Name} + om.consistencyStore.WroteAt( + ssNamespacedName, + ss.UID, + podGroupResource, + pod.ResourceVersion, + ) + } return err } @@ -88,8 +101,17 @@ func (om *realStatefulPodControlObjectManager) GetPod(namespace, podName string) return om.podLister.Pods(namespace).Get(podName) } -func (om *realStatefulPodControlObjectManager) UpdatePod(pod *v1.Pod) error { - _, err := om.client.CoreV1().Pods(pod.Namespace).Update(context.TODO(), pod, metav1.UpdateOptions{}) +func (om *realStatefulPodControlObjectManager) UpdatePod(pod *v1.Pod, ss *apps.StatefulSet) error { + pod, err := om.client.CoreV1().Pods(pod.Namespace).Update(context.TODO(), pod, metav1.UpdateOptions{}) + if err == nil { + ssNamespacedName := types.NamespacedName{Namespace: ss.Namespace, Name: ss.Name} + om.consistencyStore.WroteAt( + ssNamespacedName, + ss.UID, + podGroupResource, + pod.ResourceVersion, + ) + } return err } @@ -97,8 +119,17 @@ func (om *realStatefulPodControlObjectManager) DeletePod(pod *v1.Pod) error { return om.client.CoreV1().Pods(pod.Namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{}) } -func (om *realStatefulPodControlObjectManager) CreateClaim(claim *v1.PersistentVolumeClaim) error { - _, err := om.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Create(context.TODO(), claim, metav1.CreateOptions{}) +func (om *realStatefulPodControlObjectManager) CreateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error { + claim, err := om.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Create(context.TODO(), claim, metav1.CreateOptions{}) + if err == nil { + ssNamespacedName := types.NamespacedName{Namespace: ss.Namespace, Name: ss.Name} + om.consistencyStore.WroteAt( + ssNamespacedName, + ss.UID, + persistentVolumeClaimGroupResource, + claim.ResourceVersion, + ) + } return err } @@ -106,8 +137,17 @@ func (om *realStatefulPodControlObjectManager) GetClaim(namespace, claimName str return om.claimLister.PersistentVolumeClaims(namespace).Get(claimName) } -func (om *realStatefulPodControlObjectManager) UpdateClaim(claim *v1.PersistentVolumeClaim) error { - _, err := om.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(context.TODO(), claim, metav1.UpdateOptions{}) +func (om *realStatefulPodControlObjectManager) UpdateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error { + claim, err := om.client.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(context.TODO(), claim, metav1.UpdateOptions{}) + if err == nil { + ssNamespacedName := types.NamespacedName{Namespace: ss.Namespace, Name: ss.Name} + om.consistencyStore.WroteAt( + ssNamespacedName, + ss.UID, + persistentVolumeClaimGroupResource, + claim.ResourceVersion, + ) + } return err } @@ -118,7 +158,7 @@ func (spc *StatefulPodControl) CreateStatefulPod(ctx context.Context, set *apps. return err } // If we created the PVCs attempt to create the Pod - err := spc.objectMgr.CreatePod(ctx, pod) + err := spc.objectMgr.CreatePod(ctx, pod, set) // sink already exists errors if apierrors.IsAlreadyExists(err) { return err @@ -173,7 +213,7 @@ func (spc *StatefulPodControl) UpdateStatefulPod(ctx context.Context, set *apps. attemptedUpdate = true // commit the update, retrying on conflicts - updateErr := spc.objectMgr.UpdatePod(pod) + updateErr := spc.objectMgr.UpdatePod(pod, set) if updateErr == nil { return nil } @@ -252,7 +292,7 @@ func (spc *StatefulPodControl) UpdatePodClaimForRetentionPolicy(ctx context.Cont if !isClaimOwnerUpToDate(logger, claim, set, pod) { claim = claim.DeepCopy() // Make a copy so we don't mutate the shared cache. updateClaimOwnerRefForSetAndPod(logger, claim, set, pod) - if err := spc.objectMgr.UpdateClaim(claim); err != nil { + if err := spc.objectMgr.UpdateClaim(claim, set); err != nil { return fmt.Errorf("could not update claim %s for delete policy ownerRefs: %w", claimName, err) } } @@ -343,7 +383,7 @@ func (spc *StatefulPodControl) createPersistentVolumeClaims(set *apps.StatefulSe pvc, err := spc.objectMgr.GetClaim(claim.Namespace, claim.Name) switch { case apierrors.IsNotFound(err): - err := spc.objectMgr.CreateClaim(&claim) + err := spc.objectMgr.CreateClaim(&claim, set) if err != nil { errs = append(errs, fmt.Errorf("failed to create PVC %s: %s", claim.Name, err)) } diff --git a/pkg/controller/statefulset/stateful_pod_control_test.go b/pkg/controller/statefulset/stateful_pod_control_test.go index 4f808f18f3b..8cac11b32aa 100644 --- a/pkg/controller/statefulset/stateful_pod_control_test.go +++ b/pkg/controller/statefulset/stateful_pod_control_test.go @@ -24,11 +24,13 @@ import ( "testing" "time" + "github.com/stretchr/testify/require" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes/fake" corelisters "k8s.io/client-go/listers/core/v1" @@ -38,6 +40,7 @@ import ( "k8s.io/klog/v2/ktesting" _ "k8s.io/kubernetes/pkg/apis/apps/install" _ "k8s.io/kubernetes/pkg/apis/core/install" + "k8s.io/kubernetes/pkg/controller/util/consistency" "k8s.io/utils/ptr" ) @@ -48,7 +51,7 @@ func TestStatefulPodControlCreatesPods(t *testing.T) { fakeClient := &fake.Clientset{} claimIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) claimLister := corelisters.NewPersistentVolumeClaimLister(claimIndexer) - control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("get", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewNotFound(action.GetResource().GroupResource(), action.GetResource().Resource) }) @@ -87,7 +90,7 @@ func TestStatefulPodControlCreatePodExists(t *testing.T) { pvcIndexer.Add(&pvc) } pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { create := action.(core.CreateAction) return true, create.GetObject(), nil @@ -114,7 +117,7 @@ func TestStatefulPodControlCreatePodPvcCreateFailure(t *testing.T) { fakeClient := &fake.Clientset{} pvcIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewInternalError(errors.New("API server down")) }) @@ -149,7 +152,7 @@ func TestStatefulPodControlCreatePodPVCDeleting(t *testing.T) { pvcIndexer.Add(&pvc) } pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { create := action.(core.CreateAction) return true, create.GetObject(), nil @@ -188,7 +191,7 @@ func TestStatefulPodControlCreatePodPvcGetFailure(t *testing.T) { fakeClient := &fake.Clientset{} pvcIndexer := &fakeIndexer{getError: errors.New("API server down")} pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewInternalError(errors.New("API server down")) }) @@ -217,7 +220,7 @@ func TestStatefulPodControlCreatePodFailed(t *testing.T) { fakeClient := &fake.Clientset{} pvcIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { create := action.(core.CreateAction) return true, create.GetObject(), nil @@ -252,7 +255,7 @@ func TestStatefulPodControlNoOpUpdate(t *testing.T) { indexer.Add(&claim) } claimLister := corelisters.NewPersistentVolumeClaimLister(indexer) - control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("*", "*", func(action core.Action) (bool, runtime.Object, error) { t.Error("no-op update should not make any client invocation") return true, nil, apierrors.NewInternalError(errors.New("if we are here we have a problem")) @@ -274,7 +277,7 @@ func TestStatefulPodControlUpdatesIdentity(t *testing.T) { fakeClient := fake.NewSimpleClientset(set, pod) indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) claimLister := corelisters.NewPersistentVolumeClaimLister(indexer) - control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, claimLister, recorder, consistency.NewNoopConsistencyStore()) var updated *v1.Pod fakeClient.PrependReactor("update", "pods", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) @@ -309,7 +312,7 @@ func TestStatefulPodControlUpdateIdentityFailure(t *testing.T) { podLister := corelisters.NewPodLister(podIndexer) claimIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) claimLister := corelisters.NewPersistentVolumeClaimLister(claimIndexer) - control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "pods", func(action core.Action) (bool, runtime.Object, error) { pod.Name = "goo-0" return true, nil, apierrors.NewInternalError(errors.New("API server down")) @@ -337,7 +340,7 @@ func TestStatefulPodControlUpdatesPodStorage(t *testing.T) { fakeClient := &fake.Clientset{} pvcIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) pvcs := getPersistentVolumeClaims(set, pod) volumes := make([]v1.Volume, 0, len(pod.Spec.Volumes)) for i := range pod.Spec.Volumes { @@ -385,7 +388,7 @@ func TestStatefulPodControlUpdatePodStorageFailure(t *testing.T) { fakeClient := &fake.Clientset{} pvcIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) pvcLister := corelisters.NewPersistentVolumeClaimLister(pvcIndexer) - control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder) + control := NewStatefulPodControl(fakeClient, nil, pvcLister, recorder, consistency.NewNoopConsistencyStore()) pvcs := getPersistentVolumeClaims(set, pod) volumes := make([]v1.Volume, 0, len(pod.Spec.Volumes)) for i := range pod.Spec.Volumes { @@ -433,7 +436,7 @@ func TestStatefulPodControlUpdatePodConflictSuccess(t *testing.T) { claim := claims[k] claimIndexer.Add(&claim) } - control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistency.NewNoopConsistencyStore()) conflict := false fakeClient.AddReactor("update", "pods", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) @@ -464,7 +467,7 @@ func TestStatefulPodControlDeletesStatefulPod(t *testing.T) { set := newStatefulSet(3) pod := newStatefulSetPod(set, 0) fakeClient := &fake.Clientset{} - control := NewStatefulPodControl(fakeClient, nil, nil, recorder) + control := NewStatefulPodControl(fakeClient, nil, nil, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("delete", "pods", func(action core.Action) (bool, runtime.Object, error) { return true, nil, nil }) @@ -484,7 +487,7 @@ func TestStatefulPodControlDeleteStatefulPodNotFound(t *testing.T) { set := newStatefulSet(3) pod := newStatefulSetPod(set, 0) fakeClient := &fake.Clientset{} - control := NewStatefulPodControl(fakeClient, nil, nil, recorder) + control := NewStatefulPodControl(fakeClient, nil, nil, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("delete", "pods", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewNotFound(action.GetResource().GroupResource(), pod.Name) }) @@ -504,7 +507,7 @@ func TestStatefulPodControlDeleteFailure(t *testing.T) { set := newStatefulSet(3) pod := newStatefulSetPod(set, 0) fakeClient := &fake.Clientset{} - control := NewStatefulPodControl(fakeClient, nil, nil, recorder) + control := NewStatefulPodControl(fakeClient, nil, nil, recorder, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("delete", "pods", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewInternalError(errors.New("API server down")) }) @@ -533,7 +536,7 @@ func TestStatefulPodControlClaimsMatchDeletionPolcy(t *testing.T) { claim := claims[k] indexer.Add(&claim) } - control := NewStatefulPodControl(fakeClient, nil, claimLister, &noopRecorder{}) + control := NewStatefulPodControl(fakeClient, nil, claimLister, &noopRecorder{}, consistency.NewNoopConsistencyStore()) set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{ WhenDeleted: apps.RetainPersistentVolumeClaimRetentionPolicyType, WhenScaled: apps.RetainPersistentVolumeClaimRetentionPolicyType, @@ -598,7 +601,7 @@ func TestStatefulPodControlUpdatePodClaimForRetentionPolicy(t *testing.T) { t.Errorf("Could not add claim %s: %v", k, err) } } - control := NewStatefulPodControl(fakeClient, nil, claimLister, &noopRecorder{}) + control := NewStatefulPodControl(fakeClient, nil, claimLister, &noopRecorder{}, consistency.NewNoopConsistencyStore()) set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{ WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType, WhenScaled: apps.RetainPersistentVolumeClaimRetentionPolicyType, @@ -720,7 +723,7 @@ func TestPodClaimIsStale(t *testing.T) { pod.SetUID("123") } claimLister := corelisters.NewPersistentVolumeClaimLister(claimIndexer) - control := NewStatefulPodControl(&fake.Clientset{}, nil, claimLister, &noopRecorder{}) + control := NewStatefulPodControl(&fake.Clientset{}, nil, claimLister, &noopRecorder{}, consistency.NewNoopConsistencyStore()) expected := tc.expected // Note that the error isn't / can't be tested. if stale, _ := control.PodClaimIsStale(&set, &pod); stale != expected { @@ -758,7 +761,7 @@ func TestStatefulPodControlRetainDeletionPolicyUpdate(t *testing.T) { t.Errorf("couldn't add %s to index: %v", claim.GetName(), err) } } - control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistency.NewNoopConsistencyStore()) if err := control.UpdateStatefulPod(ctx, set, pod); err != nil { t.Errorf("Successful update returned an error: %s", err) } @@ -804,7 +807,7 @@ func TestStatefulPodControlRetentionPolicyUpdate(t *testing.T) { }) podLister := corelisters.NewPodLister(podIndexer) claimLister := corelisters.NewPersistentVolumeClaimLister(claimIndexer) - control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistency.NewNoopConsistencyStore()) if err := control.UpdateStatefulPod(ctx, set, pod); err != nil { t.Errorf("Successful update returned an error: %s", err) } @@ -842,7 +845,7 @@ func TestStatefulPodControlRetentionPolicyUpdateMissingClaims(t *testing.T) { claimIndexer.Update(update.GetObject()) return true, update.GetObject(), nil }) - control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder) + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistency.NewNoopConsistencyStore()) if err := control.UpdateStatefulPod(ctx, set, pod); err != nil { t.Error("Unexpected error on pod update when PVCs are missing") } @@ -888,3 +891,138 @@ func collectEvents(source <-chan string) []string { } return events } + +type fakeRVGetter struct { + rv string +} + +func (f *fakeRVGetter) LastStoreSyncResourceVersion() string { + return f.rv +} + +func TestStatefulPodControlConsistencyStore(t *testing.T) { + _, ctx := ktesting.NewTestContext(t) + testCases := []struct { + name string + action func(context.Context, *StatefulPodControl, *apps.StatefulSet, *v1.Pod) error + resource schema.GroupResource + setupMocks func(*fake.Clientset, cache.Indexer, *v1.Pod, *apps.StatefulSet) + }{ + { + name: "CreatePod", + action: func(ctx context.Context, control *StatefulPodControl, set *apps.StatefulSet, pod *v1.Pod) error { + return control.CreateStatefulPod(ctx, set, pod) + }, + resource: schema.GroupResource{Group: "", Resource: "pods"}, + setupMocks: func(fakeClient *fake.Clientset, claimIndexer cache.Indexer, pod *v1.Pod, set *apps.StatefulSet) { + fakeClient.AddReactor("get", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewNotFound(action.GetResource().GroupResource(), action.GetResource().Resource) + }) + fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { + create := action.(core.CreateAction) + create.GetObject().(*v1.PersistentVolumeClaim).ResourceVersion = "2" + require.NoError(t, claimIndexer.Add(create.GetObject())) + return true, create.GetObject(), nil + }) + fakeClient.AddReactor("create", "pods", func(action core.Action) (bool, runtime.Object, error) { + create := action.(core.CreateAction) + create.GetObject().(*v1.Pod).ResourceVersion = "2" + return true, create.GetObject(), nil + }) + }, + }, + { + name: "UpdatePod", + action: func(ctx context.Context, control *StatefulPodControl, set *apps.StatefulSet, pod *v1.Pod) error { + pod.Name = "goo-0" // mismatch identity to trigger update + return control.UpdateStatefulPod(ctx, set, pod) + }, + resource: schema.GroupResource{Group: "", Resource: "pods"}, + setupMocks: func(fakeClient *fake.Clientset, claimIndexer cache.Indexer, pod *v1.Pod, set *apps.StatefulSet) { + fakeClient.PrependReactor("update", "pods", func(action core.Action) (bool, runtime.Object, error) { + update := action.(core.UpdateAction) + update.GetObject().(*v1.Pod).ResourceVersion = "2" + return true, update.GetObject(), nil + }) + }, + }, + { + name: "CreateClaim", + action: func(ctx context.Context, control *StatefulPodControl, set *apps.StatefulSet, pod *v1.Pod) error { + return control.createPersistentVolumeClaims(set, pod) + }, + resource: schema.GroupResource{Group: "", Resource: "persistentvolumeclaims"}, + setupMocks: func(fakeClient *fake.Clientset, claimIndexer cache.Indexer, pod *v1.Pod, set *apps.StatefulSet) { + fakeClient.AddReactor("get", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewNotFound(action.GetResource().GroupResource(), action.GetResource().Resource) + }) + fakeClient.AddReactor("create", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { + create := action.(core.CreateAction) + create.GetObject().(*v1.PersistentVolumeClaim).ResourceVersion = "2" + require.NoError(t, claimIndexer.Add(create.GetObject())) + return true, create.GetObject(), nil + }) + }, + }, + { + name: "UpdateClaim", + action: func(ctx context.Context, control *StatefulPodControl, set *apps.StatefulSet, pod *v1.Pod) error { + return control.UpdatePodClaimForRetentionPolicy(ctx, set, pod) + }, + resource: schema.GroupResource{Group: "", Resource: "persistentvolumeclaims"}, + setupMocks: func(fakeClient *fake.Clientset, claimIndexer cache.Indexer, pod *v1.Pod, set *apps.StatefulSet) { + set.Spec.PersistentVolumeClaimRetentionPolicy = &apps.StatefulSetPersistentVolumeClaimRetentionPolicy{ + WhenDeleted: apps.DeletePersistentVolumeClaimRetentionPolicyType, + WhenScaled: apps.RetainPersistentVolumeClaimRetentionPolicyType, + } + claims := getPersistentVolumeClaims(set, pod) + for k := range claims { + claim := claims[k] + require.NoError(t, claimIndexer.Add(&claim)) + } + fakeClient.AddReactor("update", "persistentvolumeclaims", func(action core.Action) (bool, runtime.Object, error) { + update := action.(core.UpdateAction) + update.GetObject().(*v1.PersistentVolumeClaim).ResourceVersion = "2" + return true, update.GetObject(), nil + }) + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + recorder := record.NewFakeRecorder(10) + set := newStatefulSet(3) + set.GetObjectMeta().SetUID("set-123") + pod := newStatefulSetPod(set, 0) + fakeClient := &fake.Clientset{} + podIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + podLister := corelisters.NewPodLister(podIndexer) + claimIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + claimLister := corelisters.NewPersistentVolumeClaimLister(claimIndexer) + + rvGetter := &fakeRVGetter{rv: "1"} + consistencyStore := consistency.NewConsistencyStore(map[schema.GroupResource]consistency.LastSyncRVGetter{ + tc.resource: rvGetter, + }) + + control := NewStatefulPodControl(fakeClient, podLister, claimLister, recorder, consistencyStore) + + tc.setupMocks(fakeClient, claimIndexer, pod, set) + + if err := tc.action(ctx, control, set, pod); err != nil { + t.Fatalf("action failed: %v", err) + } + + if err := consistencyStore.EnsureReady(types.NamespacedName{Namespace: set.Namespace, Name: set.Name}); err == nil { + t.Error("expected consistency store to return an error, got nil") + } + + rvGetter.rv = "2" + + if err := consistencyStore.EnsureReady(types.NamespacedName{Namespace: set.Namespace, Name: set.Name}); err != nil { + t.Errorf("expected consistency store to be ready, got an error: %v", err) + } + }) + } +} diff --git a/pkg/controller/statefulset/stateful_set.go b/pkg/controller/statefulset/stateful_set.go index 2d6aaa858ae..44feceb4ba5 100644 --- a/pkg/controller/statefulset/stateful_set.go +++ b/pkg/controller/statefulset/stateful_set.go @@ -18,6 +18,7 @@ package statefulset import ( "context" + "errors" "fmt" "reflect" "sync" @@ -25,11 +26,14 @@ import ( apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" - "k8s.io/apimachinery/pkg/api/errors" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" + utilfeature "k8s.io/apiserver/pkg/util/feature" appsinformers "k8s.io/client-go/informers/apps/v1" coreinformers "k8s.io/client-go/informers/core/v1" clientset "k8s.io/client-go/kubernetes" @@ -44,6 +48,8 @@ import ( "k8s.io/kubernetes/pkg/controller" "k8s.io/kubernetes/pkg/controller/history" "k8s.io/kubernetes/pkg/controller/statefulset/metrics" + consistencyutil "k8s.io/kubernetes/pkg/controller/util/consistency" + "k8s.io/kubernetes/pkg/features" "k8s.io/utils/clock" "k8s.io/utils/ptr" @@ -84,8 +90,25 @@ type StatefulSetController struct { // eventBroadcaster is the core of event processing pipeline. eventBroadcaster record.EventBroadcaster clock clock.PassiveClock + // consistencyStore is used to track the state of the controller's operations. + consistencyStore consistencyutil.ConsistencyStore } +var ( + statefulSetGroupResource = schema.GroupResource{ + Group: "apps", + Resource: "statefulsets", + } + podGroupResource = schema.GroupResource{ + Group: "", + Resource: "pods", + } + persistentVolumeClaimGroupResource = schema.GroupResource{ + Group: "", + Resource: "persistentvolumeclaims", + } +) + // NewStatefulSetController creates a new statefulset controller. func NewStatefulSetController( ctx context.Context, @@ -104,6 +127,32 @@ func NewStatefulSetController( if err := history.AddControllerRevisionControllerIndexer(revInformer.Informer()); err != nil { utilruntime.HandleError(fmt.Errorf("unable to add controller revision controller indexer: %w", err)) } + + var consistencyStore consistencyutil.ConsistencyStore + var podWriteCallback func(pod *v1.Pod, rs *metav1.OwnerReference) + if utilfeature.DefaultFeatureGate.Enabled(features.StaleControllerConsistencyStatefulSet) { + consistencyStore = consistencyutil.NewConsistencyStore(map[schema.GroupResource]consistencyutil.LastSyncRVGetter{ + podGroupResource: podInformer.Informer().GetStore(), + persistentVolumeClaimGroupResource: pvcInformer.Informer().GetStore(), + statefulSetGroupResource: setInformer.Informer().GetStore(), + }) + podWriteCallback = func(pod *v1.Pod, ss *metav1.OwnerReference) { + if ss == nil { + return + } + consistencyStore.WroteAt( + types.NamespacedName{ + Namespace: pod.Namespace, + Name: ss.Name, + }, + ss.UID, + podGroupResource, + pod.ResourceVersion, + ) + } + } else { + consistencyStore = consistencyutil.NewNoopConsistencyStore() + } ssc := &StatefulSetController{ kubeClient: kubeClient, control: NewDefaultStatefulSetControl( @@ -111,8 +160,9 @@ func NewStatefulSetController( kubeClient, podInformer.Lister(), pvcInformer.Lister(), - recorder), - NewRealStatefulSetStatusUpdater(kubeClient, setInformer.Lister()), + recorder, + consistencyStore), + NewRealStatefulSetStatusUpdater(kubeClient, setInformer.Lister(), consistencyStore), history.NewHistory(kubeClient, revInformer.Lister(), revInformer.Informer().GetIndexer()), ), pvcListerSynced: pvcInformer.Informer().HasSynced, @@ -121,10 +171,11 @@ func NewStatefulSetController( workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: "statefulset"}, ), - podControl: controller.RealPodControl{KubeClient: kubeClient, Recorder: recorder}, + podControl: controller.RealPodControl{KubeClient: kubeClient, Recorder: recorder, OnWrite: podWriteCallback}, eventBroadcaster: eventBroadcaster, clock: clock.RealClock{}, + consistencyStore: consistencyStore, } podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ @@ -481,8 +532,25 @@ func (ssc *StatefulSetController) sync(ctx context.Context, key string) error { if err != nil { return err } + + ssNamespacedName := types.NamespacedName{ + Namespace: namespace, + Name: name, + } + if err := ssc.consistencyStore.EnsureReady(ssNamespacedName); err != nil { + var consistencyErr *consistencyutil.ConsistencyError + if errors.As(err, &consistencyErr) { + metrics.StatefulSetRequeueSkips.WithLabelValues( + consistencyErr.GroupResource.Group, + consistencyErr.GroupResource.Resource, + ).Inc() + } + return err + } + set, err := ssc.setLister.StatefulSets(namespace).Get(name) - if errors.IsNotFound(err) { + if apierrors.IsNotFound(err) { + ssc.consistencyStore.Clear(ssNamespacedName, "") logger.Info("StatefulSet has been deleted", "key", key) return nil } diff --git a/pkg/controller/statefulset/stateful_set_control_test.go b/pkg/controller/statefulset/stateful_set_control_test.go index 734572e86d4..951c8dc6423 100644 --- a/pkg/controller/statefulset/stateful_set_control_test.go +++ b/pkg/controller/statefulset/stateful_set_control_test.go @@ -2514,7 +2514,7 @@ func newFakeObjectManager(informerFactory informers.SharedInformerFactory) *fake } } -func (om *fakeObjectManager) CreatePod(ctx context.Context, pod *v1.Pod) error { +func (om *fakeObjectManager) CreatePod(ctx context.Context, pod *v1.Pod, ss *apps.StatefulSet) error { defer om.createPodTracker.trackParallelRequests() if err := om.createPodTracker.incWithOptionalError(); err != nil { return err @@ -2527,7 +2527,7 @@ func (om *fakeObjectManager) GetPod(namespace, podName string) (*v1.Pod, error) return om.podsLister.Pods(namespace).Get(podName) } -func (om *fakeObjectManager) UpdatePod(pod *v1.Pod) error { +func (om *fakeObjectManager) UpdatePod(pod *v1.Pod, ss *apps.StatefulSet) error { defer om.updatePodTracker.trackParallelRequests() if err := om.updatePodTracker.incWithOptionalError(); err != nil { return err @@ -2550,7 +2550,7 @@ func (om *fakeObjectManager) DeletePod(pod *v1.Pod) error { return nil // Not found, no error in deleting. } -func (om *fakeObjectManager) CreateClaim(claim *v1.PersistentVolumeClaim) error { +func (om *fakeObjectManager) CreateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error { om.claimsIndexer.Update(claim) return nil } @@ -2559,7 +2559,7 @@ func (om *fakeObjectManager) GetClaim(namespace, claimName string) (*v1.Persiste return om.claimsLister.PersistentVolumeClaims(namespace).Get(claimName) } -func (om *fakeObjectManager) UpdateClaim(claim *v1.PersistentVolumeClaim) error { +func (om *fakeObjectManager) UpdateClaim(claim *v1.PersistentVolumeClaim, ss *apps.StatefulSet) error { // Validate ownerRefs. refs := claim.GetOwnerReferences() for _, ref := range refs { diff --git a/pkg/controller/statefulset/stateful_set_status_updater.go b/pkg/controller/statefulset/stateful_set_status_updater.go index 409ad92e20f..5fc7577ed96 100644 --- a/pkg/controller/statefulset/stateful_set_status_updater.go +++ b/pkg/controller/statefulset/stateful_set_status_updater.go @@ -21,11 +21,13 @@ import ( apps "k8s.io/api/apps/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientset "k8s.io/client-go/kubernetes" appslisters "k8s.io/client-go/listers/apps/v1" "k8s.io/client-go/util/retry" "k8s.io/klog/v2" + consistencyutil "k8s.io/kubernetes/pkg/controller/util/consistency" ) // StatefulSetStatusUpdaterInterface is an interface used to update the StatefulSetStatus associated with a StatefulSet. @@ -40,13 +42,15 @@ type StatefulSetStatusUpdaterInterface interface { // using the supplied client and setLister. func NewRealStatefulSetStatusUpdater( client clientset.Interface, - setLister appslisters.StatefulSetLister) StatefulSetStatusUpdaterInterface { - return &realStatefulSetStatusUpdater{client, setLister} + setLister appslisters.StatefulSetLister, + consistencyStore consistencyutil.ConsistencyStore) StatefulSetStatusUpdaterInterface { + return &realStatefulSetStatusUpdater{client, setLister, consistencyStore} } type realStatefulSetStatusUpdater struct { - client clientset.Interface - setLister appslisters.StatefulSetLister + client clientset.Interface + setLister appslisters.StatefulSetLister + consistencyStore consistencyutil.ConsistencyStore } func (ssu *realStatefulSetStatusUpdater) UpdateStatefulSetStatus( @@ -58,10 +62,17 @@ func (ssu *realStatefulSetStatusUpdater) UpdateStatefulSetStatus( return retry.RetryOnConflict(retry.DefaultRetry, func() error { set.Status = *status // TODO: This context.TODO should use a real context once we have RetryOnConflictWithContext - _, updateErr := ssu.client.AppsV1().StatefulSets(set.Namespace).UpdateStatus(context.TODO(), set, metav1.UpdateOptions{}) + updatedSet, updateErr := ssu.client.AppsV1().StatefulSets(set.Namespace).UpdateStatus(context.TODO(), set, metav1.UpdateOptions{}) if updateErr == nil { + ssu.consistencyStore.WroteAt( + types.NamespacedName{Namespace: updatedSet.Namespace, Name: updatedSet.Name}, + updatedSet.UID, + statefulSetGroupResource, + updatedSet.ResourceVersion, + ) return nil } + if updated, err := ssu.setLister.StatefulSets(set.Namespace).Get(set.Name); err == nil { // make a copy so we don't mutate the shared cache set = updated.DeepCopy() diff --git a/pkg/controller/statefulset/stateful_set_status_updater_test.go b/pkg/controller/statefulset/stateful_set_status_updater_test.go index 83c51bd873a..db1f5e7384f 100644 --- a/pkg/controller/statefulset/stateful_set_status_updater_test.go +++ b/pkg/controller/statefulset/stateful_set_status_updater_test.go @@ -24,17 +24,20 @@ import ( apps "k8s.io/api/apps/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes/fake" appslisters "k8s.io/client-go/listers/apps/v1" core "k8s.io/client-go/testing" "k8s.io/client-go/tools/cache" + "k8s.io/kubernetes/pkg/controller/util/consistency" ) func TestStatefulSetUpdaterUpdatesSetStatus(t *testing.T) { set := newStatefulSet(3) status := apps.StatefulSetStatus{ObservedGeneration: 1, Replicas: 2} fakeClient := &fake.Clientset{} - updater := NewRealStatefulSetStatusUpdater(fakeClient, nil) + updater := NewRealStatefulSetStatusUpdater(fakeClient, nil, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) return true, update.GetObject(), nil @@ -51,7 +54,7 @@ func TestStatefulSetStatusUpdaterUpdatesObservedGeneration(t *testing.T) { set := newStatefulSet(3) status := apps.StatefulSetStatus{ObservedGeneration: 3, Replicas: 2} fakeClient := &fake.Clientset{} - updater := NewRealStatefulSetStatusUpdater(fakeClient, nil) + updater := NewRealStatefulSetStatusUpdater(fakeClient, nil, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) sts := update.GetObject().(*apps.StatefulSet) @@ -72,7 +75,7 @@ func TestStatefulSetStatusUpdaterUpdateReplicasFailure(t *testing.T) { indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) indexer.Add(set) setLister := appslisters.NewStatefulSetLister(indexer) - updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister) + updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { return true, nil, apierrors.NewInternalError(errors.New("API server down")) }) @@ -89,7 +92,7 @@ func TestStatefulSetStatusUpdaterUpdateReplicasConflict(t *testing.T) { indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) indexer.Add(set) setLister := appslisters.NewStatefulSetLister(indexer) - updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister) + updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) if !conflict { @@ -114,7 +117,7 @@ func TestStatefulSetStatusUpdaterUpdateReplicasConflictFailure(t *testing.T) { indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) indexer.Add(set) setLister := appslisters.NewStatefulSetLister(indexer) - updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister) + updater := NewRealStatefulSetStatusUpdater(fakeClient, setLister, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) return true, update.GetObject(), apierrors.NewConflict(action.GetResource().GroupResource(), set.Name, errors.New("object already exists")) @@ -128,7 +131,7 @@ func TestStatefulSetStatusUpdaterGetAvailableReplicas(t *testing.T) { set := newStatefulSet(3) status := apps.StatefulSetStatus{ObservedGeneration: 1, Replicas: 2, AvailableReplicas: 3} fakeClient := &fake.Clientset{} - updater := NewRealStatefulSetStatusUpdater(fakeClient, nil) + updater := NewRealStatefulSetStatusUpdater(fakeClient, nil, consistency.NewNoopConsistencyStore()) fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { update := action.(core.UpdateAction) return true, update.GetObject(), nil @@ -140,3 +143,34 @@ func TestStatefulSetStatusUpdaterGetAvailableReplicas(t *testing.T) { t.Errorf("UpdateStatefulSetStatus mutated the sets replicas %d", set.Status.AvailableReplicas) } } + +func TestStatefulSetStatusUpdaterConsistencyStore(t *testing.T) { + set := newStatefulSet(3) + status := apps.StatefulSetStatus{ObservedGeneration: 1, Replicas: 2} + fakeClient := &fake.Clientset{} + + rvGetter := &fakeRVGetter{rv: "1"} + consistencyStore := consistency.NewConsistencyStore(map[schema.GroupResource]consistency.LastSyncRVGetter{ + {Group: "apps", Resource: "statefulsets"}: rvGetter, + }) + + updater := NewRealStatefulSetStatusUpdater(fakeClient, nil, consistencyStore) + fakeClient.AddReactor("update", "statefulsets", func(action core.Action) (bool, runtime.Object, error) { + update := action.(core.UpdateAction) + update.GetObject().(*apps.StatefulSet).ResourceVersion = "2" + return true, update.GetObject(), nil + }) + if err := updater.UpdateStatefulSetStatus(context.TODO(), set, &status); err != nil { + t.Fatalf("Error returned on successful status update: %s", err) + } + + if err := consistencyStore.EnsureReady(types.NamespacedName{Namespace: set.Namespace, Name: set.Name}); err == nil { + t.Error("expected consistency store to return an error, got nil") + } + + rvGetter.rv = "2" + + if err := consistencyStore.EnsureReady(types.NamespacedName{Namespace: set.Namespace, Name: set.Name}); err != nil { + t.Errorf("expected consistency store to be ready, got an error: %v", err) + } +} diff --git a/pkg/controller/statefulset/stateful_set_test.go b/pkg/controller/statefulset/stateful_set_test.go index 40841ebe170..51340a05054 100644 --- a/pkg/controller/statefulset/stateful_set_test.go +++ b/pkg/controller/statefulset/stateful_set_test.go @@ -719,7 +719,9 @@ func TestOrphanedPodsWithPVCDeletePolicy(t *testing.T) { om.podsIndexer.Add(pod) claims := getPersistentVolumeClaims(set, pod) for _, claim := range claims { - om.CreateClaim(&claim) + if err := om.CreateClaim(&claim, set); err != nil { + t.Errorf("Failed to create claim %s: %v", claim.Name, err) + } } } diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go index 98bc2fce908..158cdee478d 100644 --- a/pkg/features/kube_features.go +++ b/pkg/features/kube_features.go @@ -948,6 +948,13 @@ const ( // prior to running a reconcile on the same object. StaleControllerConsistencyReplicaSet featuregate.Feature = "StaleControllerConsistencyReplicaSet" + // owner: @michaelasp + // kep: http://kep.k8s.io/5647 + // + // Introduces the ability for the StatefulSet controller to be able to read its writes + // prior to running a reconcile on the same object. + StaleControllerConsistencyStatefulSet featuregate.Feature = "StaleControllerConsistencyStatefulSet" + // owner: @liggitt // // Mitigates spurious statefulset rollouts due to controller revision comparison mismatches @@ -1802,6 +1809,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate {Version: version.MustParse("1.36"), Default: true, PreRelease: featuregate.Beta}, }, + StaleControllerConsistencyStatefulSet: { + {Version: version.MustParse("1.36"), Default: true, PreRelease: featuregate.Beta}, + }, + StatefulSetSemanticRevisionComparison: { // This is a mitigation for a 1.34 regression due to serialization differences that cannot be feature-gated, // so this mitigation should not auto-disable even if emulating versions prior to 1.34 with --emulation-version. @@ -2429,6 +2440,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature StaleControllerConsistencyReplicaSet: {featuregate.Feature(clientfeatures.AtomicFIFO)}, + StaleControllerConsistencyStatefulSet: {featuregate.Feature(clientfeatures.AtomicFIFO)}, + StatefulSetSemanticRevisionComparison: {}, StorageCapacityScoring: {}, diff --git a/test/compatibility_lifecycle/reference/feature_list.md b/test/compatibility_lifecycle/reference/feature_list.md index 7cab7332940..8239ae1e89f 100644 --- a/test/compatibility_lifecycle/reference/feature_list.md +++ b/test/compatibility_lifecycle/reference/feature_list.md @@ -181,6 +181,7 @@ | StaleControllerConsistencyDaemonSet | :ballot_box_with_check: 1.36+ | | | 1.36– | | | AtomicFIFO | [code](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyDaemonSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyDaemonSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | StaleControllerConsistencyJob | :ballot_box_with_check: 1.36+ | | | 1.36– | | | AtomicFIFO | [code](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyJob%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyJob%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | StaleControllerConsistencyReplicaSet | :ballot_box_with_check: 1.36+ | | | 1.36– | | | AtomicFIFO | [code](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyReplicaSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyReplicaSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | +| StaleControllerConsistencyStatefulSet | :ballot_box_with_check: 1.36+ | | | 1.36– | | | AtomicFIFO | [code](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyStatefulSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStaleControllerConsistencyStatefulSet%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | StatefulSetSemanticRevisionComparison | :ballot_box_with_check: 1.0+ | | | 1.0– | | | | [code](https://cs.k8s.io/?q=%5CbStatefulSetSemanticRevisionComparison%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStatefulSetSemanticRevisionComparison%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | StorageCapacityScoring | | | 1.33– | | | | | [code](https://cs.k8s.io/?q=%5CbStorageCapacityScoring%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStorageCapacityScoring%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | | StorageNamespaceIndex | :ballot_box_with_check: 1.30+ | | | 1.30–1.32 | | 1.33– | | [code](https://cs.k8s.io/?q=%5CbStorageNamespaceIndex%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbStorageNamespaceIndex%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) | diff --git a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml index 9f0c4545e86..e7c610be401 100644 --- a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml +++ b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml @@ -1743,6 +1743,12 @@ lockToDefault: false preRelease: Beta version: "1.36" +- name: StaleControllerConsistencyStatefulSet + versionedSpecs: + - default: true + lockToDefault: false + preRelease: Beta + version: "1.36" - name: StatefulSetSemanticRevisionComparison versionedSpecs: - default: true