Deflake Kubernetes integration tests (#69364)

Wait for Kubernetes clusters to land in the proxy's cache before
proceeding with the test.

TestKubePROXYProtocol was the most recently observed flake, but
a number of other Kube integrations tests had the same race.

Closes #69336

integration/helpers/instance.go

@@ -1407,6 +1407,10 @@ func (i *TeleInstance) Start() error {
 		return trace.Wrap(err)
 	}
 
+	if err := i.WaitForKubeClusters(context.Background()); err != nil {
+		return trace.Wrap(err)
+	}
+
 	return nil
 }
 

integration/helpers/kube.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/gravitational/trace"
 	"github.com/stretchr/testify/assert"
@@ -31,6 +32,7 @@ import (
 	apiclient "github.com/gravitational/teleport/api/client"
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/utils/retryutils"
 	"github.com/gravitational/teleport/lib/auth"
 	"github.com/gravitational/teleport/lib/kube/kubeconfig"
 	testingkubemock "github.com/gravitational/teleport/lib/kube/proxy/testing/kube_server"
@@ -85,6 +87,64 @@ func genKubeConfig(t *testing.T, kubeconfigPath, clusterName string) {
 	require.NoError(t, err)
 }
 
+// WaitForKubeClusters blocks until every kubernetes cluster in this instance's
+// kubeconfig is visible in the Proxy Service cache.
+func (i *TeleInstance) WaitForKubeClusters(ctx context.Context) error {
+	const (
+		deadline     = 30 * time.Second
+		iterWaitTime = 100 * time.Millisecond
+	)
+
+	if i.Config == nil || !i.Config.Auth.Enabled || !i.Config.Proxy.Enabled || !i.Config.Kube.Enabled {
+		return nil
+	}
+
+	// No kubeconfig means we're only supporting dynanmically registered clusters,
+	// so there's nothing to wait for.
+	if i.Config.Kube.KubeconfigPath == "" {
+		return nil
+	}
+
+	kubeConfig, err := kubeconfig.Load(i.Config.Kube.KubeconfigPath)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+	if len(kubeConfig.Contexts) == 0 {
+		return nil
+	}
+
+	err = retryutils.RetryStaticFor(deadline, iterWaitTime, func() error {
+		cluster, err := i.Tunnel.Cluster(ctx, i.Secrets.SiteName)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		accessPoint, err := cluster.CachingAccessPoint()
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		kubeServers, err := accessPoint.GetKubernetesServers(ctx)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		registered := make(map[string]struct{}, len(kubeServers))
+		for _, ks := range kubeServers {
+			registered[ks.GetCluster().GetName()] = struct{}{}
+		}
+
+		for name := range kubeConfig.Contexts {
+			if _, ok := registered[name]; !ok {
+				return trace.NotFound("kubernetes cluster %q is not registered in the proxy cache yet", name)
+			}
+		}
+
+		return nil
+	})
+	return trace.Wrap(err)
+}
+
 // GetKubeClusters gets all kubernetes clusters accessible from a given auth server.
 func GetKubeClusters(t *testing.T, as *auth.Server) []types.KubeCluster {
 	ctx := context.Background()

integration/proxy/proxy_test.go

@@ -466,7 +466,7 @@ func TestALPNSNIProxyKubeV2Leaf(t *testing.T) {
 		PinnedIP:            "127.0.0.1",
 		KubeUsers:           kubeRoleSpec.Allow.KubeGroups,
 		KubeGroups:          kubeRoleSpec.Allow.KubeUsers,
-		KubeCluster:         "gke_project_europecentral2a_cluster1",
+		KubeCluster:         kubeClusterName,
 		CustomTLSServerName: localK8SNI,
 		TargetAddress:       suite.root.Config.Proxy.WebAddr,
 		RouteToCluster:      suite.leaf.Secrets.SiteName,
@@ -610,7 +610,7 @@ func TestKubePROXYProtocol(t *testing.T) {
 				// If PROXY protocol is required, create load balancer in front of Teleport cluster
 				if tt.proxyProtocolMode == multiplexer.PROXYProtocolOn {
 					frontend := *utils.MustParseAddr("127.0.0.1:0")
-					lb, err := utils.NewLoadBalancer(context.Background(), frontend)
+					lb, err := utils.NewLoadBalancer(t.Context(), frontend)
 					require.NoError(t, err)
 					lb.PROXYHeader = []byte("PROXY TCP4 127.0.0.1 127.0.0.2 12345 42\r\n") // Send fake PROXY header
 					lb.AddBackend(targetAddr)
@@ -643,7 +643,7 @@ func TestKubePROXYProtocol(t *testing.T) {
 						kubeConfig)
 				}
 
-				resp, err := k8Client.CoreV1().Pods("default").List(context.Background(), metav1.ListOptions{})
+				resp, err := k8Client.CoreV1().Pods("default").List(t.Context(), metav1.ListOptions{})
 				require.NoError(t, err)
 				require.Len(t, resp.Items, 3, "pods item length mismatch")
 			}