Repository: kubernetes/kubernetes
File: pkg/controller/statefulset/stateful_set.go (Go)

Code before:
	// podLister is able to list/get pods from a shared informer's store
	podLister corelisters.PodLister
	// podListerSynced returns true if the pod shared informer has synced at least once
	podListerSynced cache.InformerSynced
	// setLister is able to list/get stateful sets from a shared informer's store
	setLister appslisters.StatefulSetLister
	// setListerSynced returns true if the stateful set shared informer has synced at least once
	setListerSynced cache.InformerSynced
	// pvcListerSynced returns true if the pvc shared informer has synced at least once
	pvcListerSynced cache.InformerSynced
	// revListerSynced returns true if the rev shared informer has synced at least once
	revListerSynced cache.InformerSynced
	// StatefulSets that need to be synced.
	queue workqueue.TypedRateLimitingInterface[string]
	// 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",
	}
	controllerRevisionGroupResource = schema.GroupResource{
		Group:    "apps",
		Resource: "controllerrevisions",
	}
)

// NewStatefulSetController creates a new statefulset controller.
func NewStatefulSetController(
	ctx context.Context,
	podInformer coreinformers.PodInformer,
	setInformer appsinformers.StatefulSetInformer,
	pvcInformer coreinformers.PersistentVolumeClaimInformer,
	revInformer appsinformers.ControllerRevisionInformer,
	kubeClient clientset.Interface,
) *StatefulSetController {
	logger := klog.FromContext(ctx)
	eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx))

Review comment (suggestion):
I don't see us actually recording WroteAt to this type anywhere ... don't we need to plumb consistencyStore to NewRealStatefulSetStatusUpdater?

```diff
diff --git a/pkg/controller/statefulset/stateful_set.go b/pkg/controller/statefulset/stateful_set.go
index fe96cf9d374..ce77dc31daf 100644
--- a/pkg/controller/statefulset/stateful_set.go
+++ b/pkg/controller/statefulset/stateful_set.go
@@ -167,7 +167,7 @@ func NewStatefulSetController(
 				pvcInformer.Lister(),
 				recorder,
 				consistencyStore),
-			NewRealStatefulSetStatusUpdater(kubeClient, setInformer.Lister()),
+			NewRealStatefulSetStatusUpdater(kubeClient, setInformer.Lister(), consistencyStore),
 			history.NewHistory(kubeClient, revInformer.Lister(), revInformer.Informer().GetIndexer()),
 		),
 		pvcListerSynced: pvcInformer.Informer().HasSynced,
diff --git a/pkg/controller/statefulset/stateful_set_status_updater.go b/pkg/controller/statefulset/stateful_set_status_updater.go
index 409ad92e20f..bb32288712d 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,8 +62,14 @@ 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{})
+		updatedSS, updateErr := ssu.client.AppsV1().StatefulSets(set.Namespace).UpdateStatus(context.TODO(), set, metav1.UpdateOptions{})
 		if updateErr == nil {
+			ssu.consistencyStore.WroteAt(
+				types.NamespacedName{Namespace: set.Namespace, Name: set.Name},
+				set.UID,
+				statefulSetGroupResource,
+				updatedSS.ResourceVersion,
+			)
 			return nil
 		}
 		if updated, err := ssu.setLister.StatefulSets(set.Namespace).Get(set.Name); err == nil {
```

Code after:
	// podLister is able to list/get pods from a shared informer's store
	podLister corelisters.PodLister
	// podListerSynced returns true if the pod shared informer has synced at least once
	podListerSynced cache.InformerSynced
	// setLister is able to list/get stateful sets from a shared informer's store
	setLister appslisters.StatefulSetLister
	// setListerSynced returns true if the stateful set shared informer has synced at least once
	setListerSynced cache.InformerSynced
	// pvcListerSynced returns true if the pvc shared informer has synced at least once
	pvcListerSynced cache.InformerSynced
	// revListerSynced returns true if the rev shared informer has synced at least once
	revListerSynced cache.InformerSynced
	// StatefulSets that need to be synced.
	queue workqueue.TypedRateLimitingInterface[string]
	// 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,
	podInformer coreinformers.PodInformer,
	setInformer appsinformers.StatefulSetInformer,
	pvcInformer coreinformers.PersistentVolumeClaimInformer,
	revInformer appsinformers.ControllerRevisionInformer,
	kubeClient clientset.Interface,
) *StatefulSetController {
	logger := klog.FromContext(ctx)
	eventBroadcaster := record.NewBroadcaster(record.WithContext(ctx))
	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "statefulset-controller"})

	// Register metrics
	metrics.Register()