node watcher

Author: pphelan007

Diff for: example-lb-svc.yaml

@@ -31,11 +31,15 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
 // ServiceReconciler reconciles a Service object
@@ -365,9 +369,45 @@ func (r *ServiceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		return ok && svc.Spec.Type == corev1.ServiceTypeLoadBalancer
 	})
 
+	mapNodeToLBServices := handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []reconcile.Request {
+
+		node, ok := obj.(*corev1.Node)
+		if !ok {
+			return nil
+		}
+		log.Log.Info("Node event received", "node", node.Name)
+		// List all LB Services and enqueue each
+		svcList := &corev1.ServiceList{}
+		if err := r.Client.List(ctx, svcList); err != nil {
+			// On error, return no requests
+			return nil
+		}
+
+		var reqs []reconcile.Request
+		for _, svc := range svcList.Items {
+			if svc.Spec.Type == corev1.ServiceTypeLoadBalancer {
+				reqs = append(reqs, reconcile.Request{
+					NamespacedName: types.NamespacedName{
+						Namespace: svc.Namespace,
+						Name:      svc.Name,
+					},
+				})
+			}
+		}
+		return reqs
+	},
+	)
 	return ctrl.NewControllerManagedBy(mgr).
-		For(&corev1.Service{}).
-		WithEventFilter(loadBalancerPredicate).
-		Named("service").
+		// Watch only Service objects of type=LoadBalancer
+		For(
+			&corev1.Service{},
+			builder.WithPredicates(loadBalancerPredicate),
+		).
+		// Also watch Node events, with no filter so they pass
+		Watches(
+			&corev1.Node{},
+			mapNodeToLBServices,
+		).
 		Complete(r)
+
 }

Diff for: internal/controller/service_controller.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ory/viper"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -93,6 +94,15 @@ func ParseHealthCheckOptionsFromAnnotations(annotations map[string]string) *crus
 	return healthCheckOptions
 }
 
+func isNodeReady(node *corev1.Node) bool {
+	for _, condition := range node.Status.Conditions {
+		if condition.Type == corev1.NodeReady && condition.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 func (r *ServiceReconciler) parseListenPortsAndBackends(ctx context.Context, svc *corev1.Service, logger logr.Logger) []crusoeapi.ListenPortAndBackend {
 	listenPortsAndBackends := []crusoeapi.ListenPortAndBackend{}
 
@@ -103,9 +113,17 @@ func (r *ServiceReconciler) parseListenPortsAndBackends(ctx context.Context, svc
 		return listenPortsAndBackends
 	}
 
+	// Filter out only Ready nodes
+	var readyNodes []corev1.Node
+	for _, node := range nodeList.Items {
+		if isNodeReady(&node) {
+			readyNodes = append(readyNodes, node)
+		}
+	}
+
 	// Extract internal IPs of all nodes
 	internalIPs := []string{}
-	for _, node := range nodeList.Items {
+	for _, node := range readyNodes {
 		for _, address := range node.Status.Addresses {
 			if address.Type == corev1.NodeInternalIP {
 				internalIPs = append(internalIPs, address.Address)
@@ -114,18 +132,37 @@ func (r *ServiceReconciler) parseListenPortsAndBackends(ctx context.Context, svc
 	}
 	logger.Info("Retrieved internal IPs of nodes", "internalIPs", internalIPs)
 
+	// 1. Retrieve the NodePort Service from the cluster
+	nodePortSvc := &corev1.Service{}
+	err := r.Client.Get(ctx, types.NamespacedName{Name: utils.GenerateNodePortServiceName(svc.Name), Namespace: svc.Namespace}, nodePortSvc)
+	if err != nil {
+		logger.Error(err, "Failed to get NodePort service")
+	}
+
+	// 2. Extract the NodePort from its Spec
+	var nodePorts []int32
+	for _, port := range nodePortSvc.Spec.Ports {
+		if port.NodePort != 0 {
+			nodePorts = append(nodePorts, port.NodePort)
+		}
+	}
+	logger.Info("Discovered NodePorts", "nodePorts", nodePorts)
+
 	// Map backends using the retrieved internal IPs
 	for _, port := range svc.Spec.Ports {
 		backends := []crusoeapi.Backend{}
 
-		for _, ip := range internalIPs {
-			backend := crusoeapi.Backend{
-				Ip:   ip,
-				Port: int64(port.TargetPort.IntVal), // Use the TargetPort from service spec
+		for _, nodePortVal := range nodePorts {
+			for _, ip := range internalIPs {
+				backends = append(backends, crusoeapi.Backend{
+					Ip:   ip,
+					Port: int64(nodePortVal),
+				})
 			}
-			backends = append(backends, backend)
 		}
 
+		logger.Info("Discovered backends", "backends", backends, "listenport", port.Port)
+
 		// Append to the list of ListenPortAndBackend
 		listenPortsAndBackends = append(listenPortsAndBackends, crusoeapi.ListenPortAndBackend{
 			ListenPort: int64(port.Port), // Map the port exposed by the service