debug: chain predicates to filter service updates

Author: bpdohall

internal/provider/kubernetes/controller.go

@@ -8,6 +8,7 @@ package kubernetes
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -24,7 +25,8 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-//	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -1354,32 +1356,49 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M
 	}
 
 // // create service predicate with additonal logic for update events
-// 	servicePredicateFuncs := func (filter func(*corev1.Service) bool) predicate.TypedFuncs[*corev1.Service] {
-// 		return TypedFuncs[*corev1.Service]{
-// 			CreateFunc: func(e event.TypedCreateEvent[*corev1.Service]) bool {
-// 				return filter(e.Object)
-// 			},
-// 			UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Service]) bool {
-// 				return filter(e.ObjectNew)
-// 			},
-// 			DeleteFunc: func(e event.TypedDeleteEvent[*corev1.Service]) bool {
-// 				return filter(e.Object)
-// 			},
-// 			GenericFunc: func(e event.TypedGenericEvent[*corev1.Service]) bool {
-// 				return filter(e.Object)
-// 			},
-// 		}
-// 	}
-
+	servicePredicateFuncs := predicate.TypedFuncs[*corev1.Service]{
+			CreateFunc: func(e event.TypedCreateEvent[*corev1.Service]) bool {
+				return true
+			},
+			UpdateFunc: func(e event.TypedUpdateEvent[*corev1.Service]) bool {
+				retVal := r.validateServiceUpdateForReconcile(e.ObjectOld, e.ObjectNew)
+				if strings.HasPrefix(e.ObjectOld.Name, "brendan") {
+					r.log.Info(fmt.Sprintf("predicate -- validateServiceUpdateForReconcile=%v",retVal), "name", e.ObjectOld.Name)
+				}
+				return retVal
+	
+			},
+			DeleteFunc: func(e event.TypedDeleteEvent[*corev1.Service]) bool {
+				return true
+			},
+			GenericFunc: func(e event.TypedGenericEvent[*corev1.Service]) bool {
+				return true
+			},
+		}
+	
 
 	// Watch Service CRUDs and process affected *Route objects.
-	servicePredicates := []predicate.TypedPredicate[*corev1.Service]{
+	servicePredicates := []predicate.TypedPredicate[*corev1.Service]{predicate.And[*corev1.Service](
+		servicePredicateFuncs,
 		predicate.NewTypedPredicateFuncs[*corev1.Service](func(svc *corev1.Service) bool {
-			retVal := r.validateServiceForReconcile(svc)
-			r.log.Info(fmt.Sprintf("predicate -- validateServiceForReconcile=%v",retVal), "namespace", svc.Namespace, "name", svc.Name)
-			return retVal
+			validateService := r.validateServiceForReconcile(svc)
+			if strings.HasPrefix(svc.Namespace, "dev-blue-cloud-teleportinfra-dev") {
+				r.log.Info(fmt.Sprintf("predicate -- validateServiceForReconcile=%v",validateService), "namespace", svc.Namespace, "name", svc.Name)
+			}
+			return validateService
 		}),
-	}
+	),}
+
+
+	// servicePredicates := []predicate.TypedPredicate[*corev1.Service]{
+	// 	predicate.NewTypedPredicateFuncs[*corev1.Service](func(svc *corev1.Service) bool {
+	// 		retVal := r.validateServiceForReconcile(svc)
+	// 		if strings.HasPrefix(svc.Namespace, "dev-blue-cloud-teleportinfra-dev") {
+	// 			r.log.Info(fmt.Sprintf("predicate -- validateServiceForReconcile=%v",retVal), "namespace", svc.Namespace, "name", svc.Name)
+	// 		}
+	// 		return retVal
+	// 	}),
+	// }
 
 	if r.namespaceLabel != nil {
 		servicePredicates = append(servicePredicates, predicate.NewTypedPredicateFuncs[*corev1.Service](func(svc *corev1.Service) bool {

internal/provider/kubernetes/predicates.go

@@ -280,32 +280,78 @@ func (r *gatewayAPIReconciler) isOIDCHMACSecret(nsName *types.NamespacedName) bo
 	return *nsName == oidcHMACSecret
 }
 
-// validateServiceForReconcile tries finding the owning Gateway of the Service
-// if it exists, finds the Gateway's Deployment, and further updates the Gateway
-// status Ready condition. All Services are pushed for reconciliation.
-func (r *gatewayAPIReconciler) validateServiceForReconcile(obj client.Object) bool {
+// isServiceOwnedByGateway returns true if the Service belongs to a Gateway.
+func (r *gatewayAPIReconciler) isServiceOwnedByGateway(svc *corev1.Service, updateStatus bool) bool {
 	ctx := context.Background()
-	svc, ok := obj.(*corev1.Service)
-	if !ok {
-		r.log.Info("unexpected object type, bypassing reconciliation", "object", obj)
-		return false
-	}
 	labels := svc.GetLabels()
 
 	// Check if the Service belongs to a Gateway, if so, update the Gateway status.
-	gtw := r.findOwningGateway(ctx, labels)
+	gtw := r.findOwningGateway(context.Background(), labels)
 	if gtw != nil {
-		r.updateGatewayStatus(gtw)
-		return false
+		if updateStatus {
+			r.updateGatewayStatus(gtw)
+		}
+		return true
 	}
 
 	// Merged gateways will have only this label, update status of all Gateways under found GatewayClass.
 	gcName, ok := labels[gatewayapi.OwningGatewayClassLabel]
 	if ok && r.mergeGateways.Has(gcName) {
-		if err := r.updateStatusForGatewaysUnderGatewayClass(ctx, gcName); err != nil {
-			r.log.Info("no Gateways found under GatewayClass", "name", gcName)
-			return false
+		if updateStatus {
+			if err := r.updateStatusForGatewaysUnderGatewayClass(ctx, gcName); err != nil {
+				r.log.Info("no Gateways found under GatewayClass", "name", gcName)
+				return true
+			}
 		}
+		return true
+	}
+
+	return false
+}
+
+// validateServiceUpdateForReconcile checks whether a Service update should trigger a reconcile.
+// Returns false when the backend does not have endpoint routing and the service of type clusterIP 
+// does not have a new IP address.
+func (r *gatewayAPIReconciler) validateServiceUpdateForReconcile(oldObj client.Object, newObj client.Object) bool {
+	oldSvc, ok := oldObj.(*corev1.Service)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", oldObj)
+		return true
+	}
+	newSvc, ok := newObj.(*corev1.Service)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", newObj)
+		return true
+	}
+
+	if r.isServiceOwnedByGateway(newSvc, false) {
+		return true
+	}
+
+	if (newSvc.Spec.Type != corev1.ServiceTypeClusterIP) || (oldSvc.Spec.Type != corev1.ServiceTypeClusterIP) || (newSvc.Spec.ClusterIP != oldSvc.Spec.ClusterIP) {
+		return true
+	}
+
+	nsName := utils.NamespacedName(newSvc)
+	if !r.hasRouteWithEndpointRouting(&nsName) {
+		r.log.Info("validateServiceUpdateForReconcile -- Service is not referenced by backend with endpoint routing", "service", nsName)
+		return false
+	}
+
+	return true
+}
+
+// validateServiceForReconcile tries finding the owning Gateway of the Service
+// if it exists, finds the Gateway's Deployment, and further updates the Gateway
+// status Ready condition. All Services are pushed for reconciliation.
+func (r *gatewayAPIReconciler) validateServiceForReconcile(obj client.Object) bool {
+	svc, ok := obj.(*corev1.Service)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", obj)
+		return false
+	}
+
+	if r.isServiceOwnedByGateway(svc, true) {
 		return false
 	}
 
@@ -437,6 +483,7 @@ func (r *gatewayAPIReconciler) isRouteReferencingBackend(nsName *types.Namespace
 		if len(tlsRouteList.Items) > 0 {
 			// we don't know the old value of the service clusterIP here, so we can't avoid reconciling on service change.
 			return true
+
 		}
 	}