use shared informer for node status check

Author: Trojan295

cmd/controller/run.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/castai/cluster-controller/cmd/utils"
 	"github.com/castai/cluster-controller/health"
+	"github.com/castai/cluster-controller/internal/actions"
 	"github.com/castai/cluster-controller/internal/actions/csr"
 	"github.com/castai/cluster-controller/internal/castai"
 	"github.com/castai/cluster-controller/internal/config"
@@ -131,6 +132,22 @@ func runController(
 
 	log.Infof("running castai-cluster-controller version %v, log-level: %v", binVersion, logger.Level)
 
+	// Create global informer manager if enabled
+	log.Info("initializing global informer manager...")
+	informerManager := actions.NewInformerManager(
+		log,
+		clientset,
+		cfg.Informer.ResyncPeriod,
+	)
+
+	// Start informer and wait for cache sync (blocks until synced)
+	syncCtx, syncCancel := context.WithTimeout(ctx, 30*time.Second)
+	defer syncCancel()
+
+	if err := informerManager.Start(syncCtx); err != nil {
+		return fmt.Errorf("starting informer manager: %w", err)
+	}
+
 	actionsConfig := controller.Config{
 		PollWaitInterval:     5 * time.Second,
 		PollTimeout:          maxRequestTimeout,
@@ -153,11 +170,16 @@ func runController(
 		client,
 		helmClient,
 		healthzAction,
+		informerManager,
 	)
 	defer func() {
 		if err := svc.Close(); err != nil {
 			log.Errorf("failed to close controller service: %v", err)
 		}
+		// Stop informer manager on shutdown
+		if informerManager != nil {
+			informerManager.Stop()
+		}
 	}()
 
 	if cfg.Metrics.ExportEnabled {

cmd/testserver/run.go

@@ -51,7 +51,8 @@ func run(ctx context.Context) error {
 	// Choose scenarios below by adding/removing/etc. instances of scenarios.XXX()
 	// All scenarios in the list run in parallel (but not necessarily at the same time if preparation takes different time).
 	testScenarios := []scenarios.TestScenario{
-		scenarios.CheckNodeDeletedStuck(300, logger),
+		// scenarios.CheckNodeDeletedStuck(300, logger),
+		scenarios.CheckNodeStatus(10000, logger),
 	}
 
 	var wg sync.WaitGroup

internal/actions/check_node_status.go

@@ -7,29 +7,30 @@ import (
 	"reflect"
 	"time"
 
-	"github.com/samber/lo"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/typed/core/v1"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
 
 	"github.com/castai/cluster-controller/internal/castai"
-	"github.com/castai/cluster-controller/internal/waitext"
 )
 
 var _ ActionHandler = &CheckNodeStatusHandler{}
 
-func NewCheckNodeStatusHandler(log logrus.FieldLogger, clientset kubernetes.Interface) *CheckNodeStatusHandler {
+func NewCheckNodeStatusHandler(log logrus.FieldLogger, clientset kubernetes.Interface, informerManager *InformerManager) *CheckNodeStatusHandler {
 	return &CheckNodeStatusHandler{
-		log:       log,
-		clientset: clientset,
+		log:             log,
+		clientset:       clientset,
+		informerManager: informerManager,
 	}
 }
 
 type CheckNodeStatusHandler struct {
-	log       logrus.FieldLogger
-	clientset kubernetes.Interface
+	log             logrus.FieldLogger
+	clientset       kubernetes.Interface
+	informerManager *InformerManager
 }
 
 func (h *CheckNodeStatusHandler) Handle(ctx context.Context, action *castai.ClusterAction) error {
@@ -50,7 +51,6 @@ func (h *CheckNodeStatusHandler) Handle(ctx context.Context, action *castai.Clus
 		ActionIDLogField: action.ID,
 	})
 
-	log.Info("checking status of node")
 	if req.NodeName == "" ||
 		(req.NodeID == "" && req.ProviderId == "") {
 		return fmt.Errorf("node name or node ID/provider ID is empty %w", errAction)
@@ -77,18 +77,7 @@ func (h *CheckNodeStatusHandler) checkNodeDeleted(ctx context.Context, log *logr
 	ctx, cancel := context.WithTimeout(ctx, time.Duration(timeout)*time.Second)
 	defer cancel()
 
-	b := waitext.DefaultExponentialBackoff()
-	return waitext.Retry(
-		ctx,
-		b,
-		waitext.Forever,
-		func(ctx context.Context) (bool, error) {
-			return checkNodeDeleted(ctx, h.clientset.CoreV1().Nodes(), req.NodeName, req.NodeID, req.ProviderId, log)
-		},
-		func(err error) {
-			log.Warnf("check node %s status failed, will retry: %v", req.NodeName, err)
-		},
-	)
+	return h.checkNodeDeletedWithInformer(ctx, req.NodeName, req.NodeID, req.ProviderId, log)
 }
 
 func checkNodeDeleted(ctx context.Context, clientSet v1.NodeInterface, nodeName, nodeID, providerID string, log logrus.FieldLogger) (bool, error) {
@@ -117,40 +106,167 @@ func checkNodeDeleted(ctx context.Context, clientSet v1.NodeInterface, nodeName,
 	return false, errNodeNotDeleted
 }
 
-func (h *CheckNodeStatusHandler) checkNodeReady(ctx context.Context, _ *logrus.Entry, req *castai.ActionCheckNodeStatus) error {
+func (h *CheckNodeStatusHandler) checkNodeDeletedWithInformer(ctx context.Context, nodeName, nodeID, providerID string, log logrus.FieldLogger) error {
+	lister := h.informerManager.GetNodeLister()
+
+	// Check if node is already deleted in cache
+	node, err := lister.Get(nodeName)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			log.Info("node already deleted in cache")
+			return nil // Node deleted
+		}
+		return fmt.Errorf("getting node from lister: %w", err)
+	}
+
+	// Check if node ID/provider ID don't match (name reused)
+	if err := isNodeIDProviderIDValid(node, nodeID, providerID, log); err != nil {
+		if errors.Is(err, errNodeDoesNotMatch) {
+			log.Info("node name reused, original node deleted")
+			return nil // Name reused, original deleted
+		}
+		return fmt.Errorf("validating node ID/provider ID: %w", err)
+	}
+
+	// Set up channel to receive notification when node is deleted
+	deleted := make(chan struct{})
+	errCh := make(chan error, 1)
+
+	informer := h.informerManager.GetNodeInformer()
+
+	// Register event handler to watch for node deletion
+	registration, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		UpdateFunc: func(oldObj, newObj any) {
+			node, ok := newObj.(*corev1.Node)
+			if !ok || node.Name != nodeName {
+				return
+			}
+			// Check if node was replaced (ID mismatch)
+			if err := isNodeIDProviderIDValid(node, nodeID, providerID, log); err != nil {
+				if errors.Is(err, errNodeDoesNotMatch) {
+					log.Info("node name reused, original node deleted (update event)")
+					select {
+					case deleted <- struct{}{}:
+					default:
+					}
+				}
+			}
+		},
+		DeleteFunc: func(obj any) {
+			node, ok := obj.(*corev1.Node)
+			if !ok {
+				// Handle tombstone case
+				tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					return
+				}
+				node, ok = tombstone.Obj.(*corev1.Node)
+				if !ok {
+					return
+				}
+			}
+			if node.Name == nodeName {
+				log.Info("node deleted (delete event)")
+				select {
+				case deleted <- struct{}{}:
+				default:
+				}
+			}
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to add event handler: %w", err)
+	}
+	defer func() {
+		if err := informer.RemoveEventHandler(registration); err != nil {
+			log.WithError(err).Warn("failed to remove event handler")
+		}
+	}()
+
+	// Wait for node to be deleted or timeout
+	select {
+	case <-deleted:
+		return nil
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		return fmt.Errorf("timeout waiting for node to be deleted: %w", ctx.Err())
+	}
+}
+
+func (h *CheckNodeStatusHandler) checkNodeReady(ctx context.Context, log *logrus.Entry, req *castai.ActionCheckNodeStatus) error {
+	return h.checkNodeReadyWithInformer(ctx, log, req)
+}
+
+func (h *CheckNodeStatusHandler) checkNodeReadyWithInformer(ctx context.Context, log *logrus.Entry, req *castai.ActionCheckNodeStatus) error {
 	timeout := 9 * time.Minute
 	if req.WaitTimeoutSeconds != nil {
 		timeout = time.Duration(*req.WaitTimeoutSeconds) * time.Second
 	}
 
-	watchObject := metav1.SingleObject(metav1.ObjectMeta{
-		Name: req.NodeName,
-	})
-	watchObject.TimeoutSeconds = lo.ToPtr(int64(timeout.Seconds()))
-
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
-	watch, err := h.clientset.CoreV1().Nodes().Watch(ctx, watchObject)
-	if err != nil {
-		return fmt.Errorf("creating node watch: %w", err)
+	// Check if node is already ready in cache
+	lister := h.informerManager.GetNodeLister()
+	node, err := lister.Get(req.NodeName)
+	if err == nil && h.isNodeReady(node, req.NodeID, req.ProviderId) {
+		log.Info("node already ready in cache")
+		return nil
 	}
-	defer watch.Stop()
 
-	for {
-		select {
-		case <-ctx.Done():
-			return fmt.Errorf("node %s request timeout: %v %w", req.NodeName, timeout, ctx.Err())
-		case r, ok := <-watch.ResultChan():
-			if !ok {
-				return fmt.Errorf("node %s request timeout: %v %w", req.NodeName, timeout, errNodeWatcherClosed)
+	// Set up channel to receive notification when node is ready
+	ready := make(chan struct{})
+	errCh := make(chan error, 1)
+
+	informer := h.informerManager.GetNodeInformer()
+
+	// Register event handler to watch for node updates
+	registration, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj any) {
+			node, ok := obj.(*corev1.Node)
+			if !ok || node.Name != req.NodeName {
+				return
+			}
+			if h.isNodeReady(node, req.NodeID, req.ProviderId) {
+				log.Info("node became ready (add event)")
+				select {
+				case ready <- struct{}{}:
+				default:
+				}
+			}
+		},
+		UpdateFunc: func(oldObj, newObj any) {
+			node, ok := newObj.(*corev1.Node)
+			if !ok || node.Name != req.NodeName {
+				return
 			}
-			if node, ok := r.Object.(*corev1.Node); ok {
-				if h.isNodeReady(node, req.NodeID, req.ProviderId) {
-					return nil
+			if h.isNodeReady(node, req.NodeID, req.ProviderId) {
+				log.Info("node became ready (update event)")
+				select {
+				case ready <- struct{}{}:
+				default:
 				}
 			}
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to add event handler: %w", err)
+	}
+	defer func() {
+		if err := informer.RemoveEventHandler(registration); err != nil {
+			log.WithError(err).Warn("failed to remove event handler")
 		}
+	}()
+
+	// Wait for node to be ready or timeout
+	select {
+	case <-ready:
+		return nil
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		return fmt.Errorf("timeout waiting for node to be ready: %w", ctx.Err())
 	}
 }