🐛 Fix cluster reconcilation predicates

internal/controllers/machine/machine_controller.go

@@ -115,7 +115,10 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt
 		handler.EnqueueRequestsFromMapFunc(clusterToMachines),
 		// TODO: should this wait for Cluster.Status.InfrastructureReady similar to Infra Machine resources?
 		predicates.All(ctrl.LoggerFrom(ctx),
-			predicates.ClusterUnpaused(ctrl.LoggerFrom(ctx)),
+			predicates.Any(ctrl.LoggerFrom(ctx),
+				predicates.ClusterUnpaused(ctrl.LoggerFrom(ctx)),
+				predicates.ClusterControlPlaneInitialized(ctrl.LoggerFrom(ctx)),
+			),
 			predicates.ResourceHasFilterLabel(ctrl.LoggerFrom(ctx), r.WatchFilterValue),
 		),
 	)

util/predicates/cluster_predicates.go

@@ -26,6 +26,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/util/conditions"
 )
 
 // ClusterCreateInfraReady returns a predicate that returns true for a create event when a cluster has Status.InfrastructureReady set as true
@@ -167,6 +168,45 @@ func ClusterUnpaused(logger logr.Logger) predicate.Funcs {
 	return Any(log, ClusterCreateNotPaused(log), ClusterUpdateUnpaused(log))
 }
 
+// ClusterControlPlaneInitialized returns a Predicate that returns true on Update events
+// when ControlPlaneInitializedCondition on a Cluster changes to true.
+// Example use:
+//  err := controller.Watch(
+//      &source.Kind{Type: &clusterv1.Cluster{}},
+//      &handler.EnqueueRequestsFromMapFunc{
+//          ToRequests: clusterToMachines,
+//      },
+//      predicates.ClusterControlPlaneInitialized(r.Log),
+//  )
+func ClusterControlPlaneInitialized(logger logr.Logger) predicate.Funcs {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			log := logger.WithValues("predicate", "ClusterControlPlaneInitialized", "eventType", "update")
+
+			oldCluster, ok := e.ObjectOld.(*clusterv1.Cluster)
+			if !ok {
+				log.V(4).Info("Expected Cluster", "type", fmt.Sprintf("%T", e.ObjectOld))
+				return false
+			}
+			log = log.WithValues("namespace", oldCluster.Namespace, "cluster", oldCluster.Name)
+
+			newCluster := e.ObjectNew.(*clusterv1.Cluster)
+
+			if !conditions.IsTrue(oldCluster, clusterv1.ControlPlaneInitializedCondition) &&
+				conditions.IsTrue(newCluster, clusterv1.ControlPlaneInitializedCondition) {
+				log.V(6).Info("Cluster ControlPlaneInitialized was set, allow further processing")
+				return true
+			}
+
+			log.V(6).Info("Cluster ControlPlaneInitialized hasn't changed, blocking further processing")
+			return false
+		},
+		CreateFunc:  func(e event.CreateEvent) bool { return false },
+		DeleteFunc:  func(e event.DeleteEvent) bool { return false },
+		GenericFunc: func(e event.GenericEvent) bool { return false },
+	}
+}
+
 // ClusterUnpausedAndInfrastructureReady returns a Predicate that returns true on Cluster creation events where
 // both Cluster.Spec.Paused is false and Cluster.Status.InfrastructureReady is true and Update events when
 // either Cluster.Spec.Paused transitions to false or Cluster.Status.InfrastructureReady transitions to true.

util/predicates/cluster_predicates_test.go

@@ -0,0 +1,98 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package predicates_test
+
+import (
+	"testing"
+
+	"github.com/go-logr/logr"
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	"sigs.k8s.io/cluster-api/util/conditions"
+	"sigs.k8s.io/cluster-api/util/predicates"
+)
+
+func TestClusterControlplaneInitializedPredicate(t *testing.T) {
+	g := NewWithT(t)
+	predicate := predicates.ClusterControlPlaneInitialized(logr.New(log.NullLogSink{}))
+
+	markedFalse := clusterv1.Cluster{}
+	conditions.MarkFalse(&markedFalse, clusterv1.ControlPlaneInitializedCondition, clusterv1.MissingNodeRefReason, clusterv1.ConditionSeverityWarning, "")
+
+	markedTrue := clusterv1.Cluster{}
+	conditions.MarkTrue(&markedTrue, clusterv1.ControlPlaneInitializedCondition)
+
+	notMarked := clusterv1.Cluster{}
+
+	testcases := []struct {
+		name       string
+		oldCluster clusterv1.Cluster
+		newCluster clusterv1.Cluster
+		expected   bool
+	}{
+		{
+			name:       "no conditions -> no conditions: should return false",
+			oldCluster: notMarked,
+			newCluster: notMarked,
+			expected:   false,
+		},
+		{
+			name:       "no conditions -> true: should return true",
+			oldCluster: notMarked,
+			newCluster: markedTrue,
+			expected:   true,
+		},
+		{
+			name:       "false -> true: should return true",
+			oldCluster: markedFalse,
+			newCluster: markedTrue,
+			expected:   true,
+		},
+		{
+			name:       "no conditions -> false: should return false",
+			oldCluster: notMarked,
+			newCluster: markedFalse,
+			expected:   false,
+		},
+		{
+			name:       "true -> false: should return false",
+			oldCluster: markedTrue,
+			newCluster: markedFalse,
+			expected:   false,
+		},
+		{
+			name:       "true -> true: should return false",
+			oldCluster: markedTrue,
+			newCluster: markedTrue,
+			expected:   false,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			ev := event.UpdateEvent{
+				ObjectOld: &tc.oldCluster,
+				ObjectNew: &tc.newCluster,
+			}
+
+			g.Expect(predicate.Update(ev)).To(Equal(tc.expected))
+		})
+	}
+}