Merge pull request #2182 from shiftstack/watch

Reduce reconciles and logs

Author: k8s-ci-robot

controllers/openstackcluster_controller.go

@@ -865,6 +865,11 @@ func (r *OpenStackClusterReconciler) SetupWithManager(ctx context.Context, mgr c
 			}),
 			builder.WithPredicates(predicates.ClusterUnpaused(ctrl.LoggerFrom(ctx))),
 		).
+		Watches(
+			&infrav1alpha1.OpenStackServer{},
+			handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &infrav1.OpenStackCluster{}),
+			builder.WithPredicates(OpenStackServerReconcileComplete(log)),
+		).
 		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), r.WatchFilterValue)).
 		WithEventFilter(predicates.ResourceIsNotExternallyManaged(ctrl.LoggerFrom(ctx))).
 		Complete(r)

controllers/openstackmachine_controller.go

@@ -188,35 +188,35 @@ func patchMachine(ctx context.Context, patchHelper *patch.Helper, openStackMachi
 }
 
 func (r *OpenStackMachineReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
+	log := ctrl.LoggerFrom(ctx)
+
 	return ctrl.NewControllerManagedBy(mgr).
 		WithOptions(options).
 		For(&infrav1.OpenStackMachine{}).
 		Watches(
 			&clusterv1.Machine{},
 			handler.EnqueueRequestsFromMapFunc(util.MachineToInfrastructureMapFunc(infrav1.SchemeGroupVersion.WithKind("OpenStackMachine"))),
 		).
-		Watches(
-			&infrav1.OpenStackCluster{},
-			handler.EnqueueRequestsFromMapFunc(r.OpenStackClusterToOpenStackMachines(ctx)),
-		).
 		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), r.WatchFilterValue)).
 		Watches(
 			&clusterv1.Cluster{},
 			handler.EnqueueRequestsFromMapFunc(r.requeueOpenStackMachinesForUnpausedCluster(ctx)),
-			builder.WithPredicates(predicates.ClusterUnpausedAndInfrastructureReady(ctrl.LoggerFrom(ctx))),
+			builder.WithPredicates(predicates.ClusterUnpausedAndInfrastructureReady(log)),
 		).
+		// NOTE: we don't watch OpenStackCluster here, even though the
+		// OpenStackMachine controller directly requires values from
+		// OpenStackCluster. The reason is that we are already observing Cluster
+		// with the ClusterUnpausedAndInfrastructureReady predicate. The only
+		// fields in OpenStackCluster we are interested in are dependent on
+		// InfrastructureReady, so we don't need to watch both.
 		Watches(
 			&ipamv1.IPAddressClaim{},
 			handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &infrav1.OpenStackMachine{}),
 		).
-		// TODO(emilien) to optimize because it's not efficient to watch all OpenStackServer events.
-		// We are only interested in certain state transitions of the OpenStackServer:
-		// - when the server is deleted
-		// - when the server becomes ready
-		// For that we probably want to write Predicate functions for the OpenStackServer.
 		Watches(
 			&infrav1alpha1.OpenStackServer{},
 			handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &infrav1.OpenStackMachine{}),
+			builder.WithPredicates(OpenStackServerReconcileComplete(log)),
 		).
 		Complete(r)
 }

controllers/openstackserver_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	ipamv1 "sigs.k8s.io/cluster-api/exp/ipam/api/v1beta1"
@@ -42,7 +43,9 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	orcv1alpha1 "github.com/k-orc/openstack-resource-controller/api/v1alpha1"
@@ -285,9 +288,16 @@ func objToDependencyLabel(obj client.Object, scheme *runtime.Scheme) (string, er
 	return "orc-dependency-" + prefixType + "/" + obj.GetName(), nil
 }
 
+func IsServerTerminalError(server *infrav1alpha1.OpenStackServer) bool {
+	if server.Status.InstanceState != nil && *server.Status.InstanceState == infrav1.InstanceStateError {
+		return true
+	}
+	return false
+}
+
 func (r *OpenStackServerReconciler) reconcileNormal(ctx context.Context, scope *scope.WithLogger, openStackServer *infrav1alpha1.OpenStackServer) (_ ctrl.Result, reterr error) {
 	// If the OpenStackServer is in an error state, return early.
-	if openStackServer.Status.InstanceState != nil && *openStackServer.Status.InstanceState == infrav1.InstanceStateError {
+	if IsServerTerminalError(openStackServer) {
 		scope.Logger().Info("Not reconciling server in error state. See openStackServer.status or previously logged error for details")
 		return ctrl.Result{}, nil
 	}
@@ -708,3 +718,56 @@ func (r *OpenStackServerReconciler) reconcileDeleteFloatingAddressFromPool(scope
 	controllerutil.RemoveFinalizer(claim, infrav1.IPClaimMachineFinalizer)
 	return r.Client.Update(context.Background(), claim)
 }
+
+// OpenStackServerReconcileComplete returns a predicate that determines if a OpenStackServer has finished reconciling.
+func OpenStackServerReconcileComplete(log logr.Logger) predicate.Funcs {
+	log = log.WithValues("predicate", "OpenStackServerReconcileComplete")
+
+	return predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool {
+			log = log.WithValues("eventType", "create")
+
+			server, ok := e.Object.(*infrav1alpha1.OpenStackServer)
+			if !ok {
+				log.V(4).Info("Expected OpenStackServer", "type", fmt.Sprintf("%T", e.Object))
+				return false
+			}
+			log = log.WithValues("OpenStackServer", klog.KObj(server))
+
+			if server.Status.Ready || IsServerTerminalError(server) {
+				log.V(6).Info("OpenStackServer finished reconciling, allowing further processing")
+				return true
+			}
+			log.V(6).Info("OpenStackServer is still reconciling, blocking further processing")
+			return false
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			log := log.WithValues("eventType", "update")
+
+			oldServer, ok := e.ObjectOld.(*infrav1alpha1.OpenStackServer)
+			if !ok {
+				log.V(4).Info("Expected OpenStackServer", "type", fmt.Sprintf("%T", e.ObjectOld))
+				return false
+			}
+			log = log.WithValues("OpenStackServer", klog.KObj(oldServer))
+
+			newServer, ok := e.ObjectNew.(*infrav1alpha1.OpenStackServer)
+			if !ok {
+				log.V(4).Info("Expected OpenStackServer (new)", "type", fmt.Sprintf("%T", e.ObjectNew))
+				return false
+			}
+
+			oldFinished := oldServer.Status.Ready || IsServerTerminalError(oldServer)
+			newFinished := newServer.Status.Ready || IsServerTerminalError(newServer)
+			if !oldFinished && newFinished {
+				log.V(6).Info("OpenStackServer finished reconciling, allowing further processing")
+				return true
+			}
+
+			log.V(4).Info("OpenStackServer is still reconciling, blocking further processing")
+			return false
+		},
+		DeleteFunc:  func(event.DeleteEvent) bool { return false },
+		GenericFunc: func(event.GenericEvent) bool { return false },
+	}
+}

pkg/cloud/services/networking/securitygroups.go

@@ -443,24 +443,28 @@ func (s *Service) reconcileGroupRules(desired *securityGroupSpec, observed *grou
 		}
 	}
 
-	s.scope.Logger().V(4).Info("Deleting rules not needed anymore for group", "name", observed.Name, "amount", len(rulesToDelete))
-	for _, rule := range rulesToDelete {
-		s.scope.Logger().V(6).Info("Deleting rule", "ID", rule, "name", observed.Name)
-		err := s.client.DeleteSecGroupRule(rule)
-		if err != nil {
-			return err
+	if len(rulesToDelete) > 0 {
+		s.scope.Logger().V(4).Info("Deleting rules not needed anymore for group", "name", observed.Name, "amount", len(rulesToDelete))
+		for _, rule := range rulesToDelete {
+			s.scope.Logger().V(6).Info("Deleting rule", "ID", rule, "name", observed.Name)
+			err := s.client.DeleteSecGroupRule(rule)
+			if err != nil {
+				return err
+			}
 		}
 	}
 
-	s.scope.Logger().V(4).Info("Creating new rules needed for group", "name", observed.Name, "amount", len(rulesToCreate))
-	for _, rule := range rulesToCreate {
-		r := rule
-		if r.RemoteGroupID == remoteGroupIDSelf {
-			r.RemoteGroupID = observed.ID
-		}
-		err := s.createRule(observed.ID, r)
-		if err != nil {
-			return err
+	if len(rulesToCreate) > 0 {
+		s.scope.Logger().V(4).Info("Creating new rules needed for group", "name", observed.Name, "amount", len(rulesToCreate))
+		for _, rule := range rulesToCreate {
+			r := rule
+			if r.RemoteGroupID == remoteGroupIDSelf {
+				r.RemoteGroupID = observed.ID
+			}
+			err := s.createRule(observed.ID, r)
+			if err != nil {
+				return err
+			}
 		}
 	}