operator v1: use uncached client in initial Get Cluster

It's possible that Get returns a cached, stale value - even if the
missed change was performed in the same process. This is because of
various trade-offs made in controller-runtime's cache layers.

This brings the controller in a dangerous situtation, in practice it
happens that a previous run reduced status.currentReplicas from 3 to 2,
but the next run reads a stale object and sees 3.

To avoid this bug, and other bugs we can not yet foresee, we will
perform this one Get with an uncached read. It is a trade-off, where we
accept a performance hit, and increased load on kube-api server, but we
value the consistency higher. This is limited to this one, most
important Get call, of Cluster resource only (one per entire Reconcile).

Since controller-runtime has throttling built in, and it's only "that
on Cluster Get call", we believe this trade-off is acceptable, and the
benefits outweight the cost.

Author: birdayz

operator/cmd/run/run.go

@@ -27,6 +27,7 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrlClient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -252,8 +253,15 @@ func Run(
 
 		adminAPIClientFactory := adminutils.CachedNodePoolAdminAPIClientFactory(adminutils.NewNodePoolInternalAdminAPI)
 
+		restConfig := mgr.GetConfig()
+		uncachedClient, err := ctrlClient.New(restConfig, ctrlClient.Options{Scheme: mgr.GetScheme(), Mapper: mgr.GetRESTMapper()})
+		if err != nil {
+			return fmt.Errorf("failed to create uncached client: %w", err)
+		}
+
 		if err = (&vectorizedcontrollers.ClusterReconciler{
 			Client:                    mgr.GetClient(),
+			UncachedClient:            uncachedClient,
 			Log:                       ctrl.Log.WithName("controllers").WithName("redpanda").WithName("Cluster"),
 			Scheme:                    mgr.GetScheme(),
 			AdminAPIClientFactory:     adminutils.NewNodePoolInternalAdminAPI,

operator/internal/controller/vectorized/cluster_controller.go

@@ -69,6 +69,7 @@ var (
 // ClusterReconciler reconciles a Cluster object
 type ClusterReconciler struct {
 	client.Client
+	UncachedClient            client.Client
 	Log                       logr.Logger
 	configuratorSettings      resources.ConfiguratorSettings
 	clusterDomain             string
@@ -124,17 +125,25 @@ func (r *ClusterReconciler) Reconcile(
 
 	var vectorizedCluster vectorizedv1alpha1.Cluster
 	ar := newAttachedResources(ctx, r, log, &vectorizedCluster)
-	if err := r.Get(ctx, req.NamespacedName, &vectorizedCluster); err != nil {
-		// we'll ignore not-found errors, since they can't be fixed by an immediate
-		// requeue (we'll need to wait for a new notification), and we can get them
-		// on deleted requests.
-		if apierrors.IsNotFound(err) {
-			if removeError := ar.getClusterRoleBinding().RemoveSubject(ctx, req.NamespacedName); removeError != nil {
-				return ctrl.Result{}, fmt.Errorf("unable to remove subject in ClusterroleBinding: %w", removeError)
+	{
+		// Perform Get with uncached client, to avoid cache races, where we don't
+		// see previous changes performed via Patch/Update.
+		getClient := r.Client
+		if r.UncachedClient != nil {
+			getClient = r.UncachedClient
+		}
+		if err := getClient.Get(ctx, req.NamespacedName, &vectorizedCluster); err != nil {
+			// we'll ignore not-found errors, since they can't be fixed by an immediate
+			// requeue (we'll need to wait for a new notification), and we can get them
+			// on deleted requests.
+			if apierrors.IsNotFound(err) {
+				if removeError := ar.getClusterRoleBinding().RemoveSubject(ctx, req.NamespacedName); removeError != nil {
+					return ctrl.Result{}, fmt.Errorf("unable to remove subject in ClusterroleBinding: %w", removeError)
+				}
+				return ctrl.Result{}, nil
 			}
-			return ctrl.Result{}, nil
+			return ctrl.Result{}, fmt.Errorf("unable to retrieve Cluster resource: %w", err)
 		}
-		return ctrl.Result{}, fmt.Errorf("unable to retrieve Cluster resource: %w", err)
 	}
 
 	// After every reconciliation, update status: