github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 11:28:58 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #137254 from michaelasp/statefulConsistency

Browse Source 
Add the ability for the statefulset controller to read its own writes
This commit is contained in:
Kubernetes Prow Robot
2026-02-28 01:39:30 +05:30
committed by GitHub
parent fe943aff0c c8e8bd5085
commit 330950ca52
11 changed files with 386 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
pkg/controller/statefulset/metrics/metrics.go
View File

@@ -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

80
pkg/controller/statefulset/stateful_pod_control.go
View File

@@ -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))
}

180
pkg/controller/statefulset/stateful_pod_control_test.go
View File

@@ -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)
}
})
}
}

78
pkg/controller/statefulset/stateful_set.go
View File

@@ -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
}

8
pkg/controller/statefulset/stateful_set_control_test.go
View File

@@ -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 {

21
pkg/controller/statefulset/stateful_set_status_updater.go
View File

@@ -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()

46
pkg/controller/statefulset/stateful_set_status_updater_test.go
View File

@@ -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)
}
}

4
pkg/controller/statefulset/stateful_set_test.go
View File

@@ -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)
}
}
}

13
pkg/features/kube_features.go
View File

@@ -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.
@@ -2430,6 +2441,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature
StaleControllerConsistencyReplicaSet: {featuregate.Feature(clientfeatures.AtomicFIFO)},
StaleControllerConsistencyStatefulSet: {featuregate.Feature(clientfeatures.AtomicFIFO)},
StatefulSetSemanticRevisionComparison: {},
StorageCapacityScoring: {},

1
test/compatibility_lifecycle/reference/feature_list.md
View File

@@ -181,6 +181,7 @@
| StaleControllerConsistencyDaemonSet | :ballot_box_with_check:&nbsp;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:&nbsp;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:&nbsp;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:&nbsp;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:&nbsp;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:&nbsp;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) |

6
test/compatibility_lifecycle/reference/versioned_feature_list.yaml
View File

@@ -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