fix(e2e): make WaitUntilNodeReady robust to watch disconnections

Replace watch-based node waiting with ticker-based polling.

- Use List instead of Watch to avoid silent failures on network errors
- Return (string, error) instead of calling t.Fatalf()
- Log transient API errors for debugging
- Provide clear error messages distinguishing "not found" vs "not ready"

Author: r2k1

e2e/kube.go

@@ -6,20 +6,17 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
-	"testing"
 	"time"
 
 	"github.com/Azure/agentbaker/e2e/config"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	errorsk8s "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
@@ -143,60 +140,51 @@ func (k *Kubeclient) WaitUntilPodRunning(ctx context.Context, namespace string,
 	return k.WaitUntilPodRunningWithRetry(ctx, namespace, labelSelector, fieldSelector, 0)
 }
 
-func (k *Kubeclient) WaitUntilNodeReady(ctx context.Context, t testing.TB, vmssName string) string {
-	startTime := time.Now()
-	t.Logf("waiting for node %s to be ready in k8s API", vmssName)
-	defer func() {
-		t.Logf("waited for node %s to be ready in k8s API for %s", vmssName, time.Since(startTime))
-	}()
-
-	var node *corev1.Node = nil
-	watcher, err := k.Typed.CoreV1().Nodes().Watch(ctx, metav1.ListOptions{})
-	require.NoError(t, err, "failed to start watching nodes")
-	defer watcher.Stop()
-
-	for event := range watcher.ResultChan() {
-		if event.Type != watch.Added && event.Type != watch.Modified {
-			continue
-		}
-
-		var nodeFromEvent *corev1.Node
-		switch v := event.Object.(type) {
-		case *corev1.Node:
-			nodeFromEvent = v
+func (k *Kubeclient) WaitUntilNodeReady(ctx context.Context, vmssName string) (string, error) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
 
-		default:
-			t.Logf("skipping object type %T", event.Object)
-			continue
-		}
+	var foundNode *corev1.Node
 
-		if !strings.HasPrefix(nodeFromEvent.Name, vmssName) {
-			continue
+	for {
+		nodes, err := k.Typed.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+		if err != nil {
+			logf(ctx, "failed to list nodes: %v", err)
+		} else {
+			foundNode = nil
+			for _, node := range nodes.Items {
+				if strings.HasPrefix(node.Name, vmssName) {
+					foundNode = &node
+					if isNodeReady(&node) {
+						return node.Name, nil
+					}
+					break
+				}
+			}
 		}
 
-		// found the right node. Use it!
-		node = nodeFromEvent
-		nodeTaints, _ := json.Marshal(node.Spec.Taints)
-		nodeConditions, _ := json.Marshal(node.Status.Conditions)
-
-		for _, cond := range node.Status.Conditions {
-			if cond.Type == corev1.NodeReady && cond.Status == corev1.ConditionTrue {
-				t.Logf("node %s is ready. Taints: %s Conditions: %s", node.Name, string(nodeTaints), string(nodeConditions))
-				return node.Name
+		select {
+		case <-ctx.Done():
+			if foundNode != nil {
+				return "", fmt.Errorf("node %s not ready: conditions=%v, taints=%v: %w",
+					foundNode.Name,
+					foundNode.Status.Conditions,
+					foundNode.Spec.Taints,
+					ctx.Err())
 			}
+			return "", fmt.Errorf("node with prefix %q not found: %w", vmssName, ctx.Err())
+		case <-ticker.C:
 		}
-
-		t.Logf("node %s is not ready. Taints: %s Conditions: %s", node.Name, string(nodeTaints), string(nodeConditions))
 	}
+}
 
-	if node == nil {
-		t.Fatalf("%q haven't appeared in k8s API server", vmssName)
-		return ""
+func isNodeReady(node *corev1.Node) bool {
+	for _, cond := range node.Status.Conditions {
+		if cond.Type == corev1.NodeReady && cond.Status == corev1.ConditionTrue {
+			return true
+		}
 	}
-
-	nodeString, _ := json.Marshal(node)
-	t.Fatalf("failed to wait for %q (%s) to be ready %+v. Detail: %s", vmssName, node.Name, node.Status, string(nodeString))
-	return node.Name
+	return false
 }
 
 // GetPodNetworkDebugPodForNode returns a pod that's a member of the 'debugnonhost' daemonset running in the cluster - this will return

e2e/test_helpers.go

@@ -280,12 +280,13 @@ func prepareAKSNode(ctx context.Context, s *Scenario) (*ScenarioVM, error) {
 	require.NoError(s.T, err)
 
 	if !s.Config.SkipDefaultValidation {
-		vmssCreatedAt := time.Now()         // Record the start time
-		creationElapse := time.Since(start) // Calculate the elapsed time
-		scenarioVM.KubeName = s.Runtime.Cluster.Kube.WaitUntilNodeReady(ctx, s.T, s.Runtime.VMSSName)
-		readyElapse := time.Since(vmssCreatedAt) // Calculate the elapsed time
+		vmssCreatedAt := time.Now()
+		kubeName, err := s.Runtime.Cluster.Kube.WaitUntilNodeReady(ctx, s.Runtime.VMSSName)
+		require.NoError(s.T, err)
+		scenarioVM.KubeName = kubeName
+		nodeReadyElapse := time.Since(vmssCreatedAt)
 		totalElapse := time.Since(start)
-		toolkit.LogDuration(ctx, totalElapse, 3*time.Minute, fmt.Sprintf("Node %s took %s to be created and %s to be ready", s.Runtime.VMSSName, creationElapse, readyElapse))
+		toolkit.LogDuration(ctx, totalElapse, 3*time.Minute, fmt.Sprintf("Node %s: total %s (VMSS creation + %s for node ready)", s.Runtime.VMSSName, totalElapse, nodeReadyElapse))
 	}
 
 	return scenarioVM, nil