Apply()
	if err != nil {
		return
	}

	// Wait for redis and ratelimit service to be up.
	fetchFn := kube.NewPodFetch(ctx.Clusters().Default(), ratelimitNs.Name(), "app=redis")
	if _, err = kube.WaitUntilPodsAreReady(fetchFn); err != nil {
		return
	}
	fetchFn = kube.NewPodFetch(ctx.Clusters().Default(), ratelimitNs.Name(), "app=ratelimit")
	if _, err = kube.WaitUntilPodsAreReady(fetchFn); err != nil {
		return
	}

---
`).ApplyOrFail(t, apply.NoCleanup)
			cs := t.Clusters().Default().(*kubecluster.Cluster)
			retry.UntilSuccessOrFail(t, func() error {
				_, err := kubetest.CheckPodsAreReady(kubetest.NewPodFetch(cs, gatewayNs.Name(), "istio=ingressgateway"))
				return err
			}, retry.Timeout(time.Minute*2), retry.Delay(time.Second))

	ErrNoPodsFetched = fmt.Errorf("no pods fetched")
)

// PodFetchFunc fetches pods from a k8s Client.
type PodFetchFunc func() ([]corev1.Pod, error)

// NewPodFetch creates a new PodFetchFunction that fetches all pods matching the namespace and label selectors.
func NewPodFetch(a istioKube.CLIClient, namespace string, selectors ...string) PodFetchFunc {
	return func() ([]corev1.Pod, error) {

		return nil, fmt.Errorf("failed to apply rendered %s, err: %v", env.RegistryRedirectorServerInstallFilePath, err)
	}

	fetchFn := testKube.NewPodFetch(ctx.Clusters().Default(), c.ns.Name(), podSelector)
	pods, err := testKube.WaitUntilPodsAreReady(fetchFn)
	if err != nil {
		return nil, err
	}

        port:
          number: 80
`).Apply(apply.NoCleanup)
				cs := t.Clusters().Default().(*kubecluster.Cluster)
				retry.UntilSuccessOrFail(t, func() error {
					_, err := kubetest.CheckPodsAreReady(kubetest.NewPodFetch(cs, gatewayNs.Name(), "istio=custom"))
					return err
				}, retry.Timeout(time.Minute*2))
				apps.B[0].CallOrFail(t, echo.CallOptions{
					Port:    echo.Port{ServicePort: 80},
					Scheme:  scheme.HTTP,

		return err
	}

	// Wait for the endpoints to be ready.
	var g multierror.Group
	for _, c := range ctx.AllClusters() {
		c := c
		g.Go(func() error {
			fetchFn := kube.NewPodFetch(c, s.ns.Name(), "app=jwt-server")
			_, err := kube.WaitUntilPodsAreReady(fetchFn)
			if err != nil {
				return fmt.Errorf("jwt-server pod not ready in cluster %s: %v", c.Name(), err)
			}

		sanitycheck.RunTrafficTestClientServer(t, oldClient, newServer)
	}
}

func checkVersion(t framework.TestContext, namespace, version string) error {
	// func NewPodFetch(a istioKube.CLIClient, namespace string, selectors ...string) PodFetchFunc {
	fetch := kubetest.NewPodFetch(t.Clusters().Default(), namespace)
	pods, err := kubetest.CheckPodsAreReady(fetch)
	if err != nil {
		return fmt.Errorf("failed to retrieve pods: %v", err)
	}

			sa := apps.Captured[0].ServiceAccountName()

			// we do not know which ztunnel instance is located on the node as the workload, so we need to check all of them initially
			ztunnelPods, err := kubetest.NewPodFetch(t.AllClusters()[0], istioCfg.SystemNamespace, "app=ztunnel")()
			assert.NoError(t, err)

			originalWorkloadSecret, ztunnelPod, err := getWorkloadSecret(t, ztunnelPods, sa, istioCtl)
			if err != nil {

		"waypoint",
		"delete",
		"--namespace",
		ns.Name(),
		waypoint,
	})
	waypointError := retry.UntilSuccess(func() error {
		fetch := testKube.NewPodFetch(t.AllClusters()[0], ns.Name(), constants.GatewayNameLabel+"="+waypoint)
		pods, err := testKube.CheckPodsAreReady(fetch)
		if err != nil && !errors.Is(err, testKube.ErrNoPodsFetched) {

	}
)

func deploy(t framework.TestContext, name, ns, yaml string) {
	t.ConfigIstio().File(ns, yaml).ApplyOrFail(t)
	if _, err := kube.WaitUntilPodsAreReady(kube.NewPodFetch(t.Clusters().Default(), ns, "app="+name)); err != nil {
		t.Fatalf("Wait for pod %s failed: %v", name, err)
	}
	t.Logf("deploy %s is ready", name)
}

func waitService(t framework.TestContext, name, ns string) {