test/e2e/framework/k8s.go - apisix-ingress-controller - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.

 package framework

 import (
 	"bufio"
 	"bytes"
 	"cmp"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"strings"
 	"sync"
 	"time"

 	"github.com/api7/gopkg/pkg/log"
 	"github.com/gavv/httpexpect/v2"
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/gruntwork-io/terratest/modules/testing"
 	. "github.com/onsi/gomega" //nolint:staticcheck
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
 	applymetav1 "k8s.io/client-go/applyconfigurations/meta/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/remotecommand"
 	"k8s.io/kubectl/pkg/scheme"
 	"k8s.io/utils/ptr"
 )

 // buildRestConfig builds the rest.Config object from kubeconfig filepath and
 // context, if kubeconfig is missing, building from in-cluster configuration.
 func buildRestConfig(context string) (*rest.Config, error) {

 	// Config loading rules:
 	// 1. kubeconfig if it not empty string
 	// 2. Config(s) in KUBECONFIG environment variable
 	// 3. In cluster config if running in-cluster
 	// 4. Use $HOME/.kube/config
 	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
 	loadingRules.DefaultClientConfig = &clientcmd.DefaultClientConfig
 	configOverrides := &clientcmd.ConfigOverrides{
 		ClusterDefaults: clientcmd.ClusterDefaults,
 		CurrentContext:  context,
 	}

 	clientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
 	return clientConfig.ClientConfig()
 }

 func (f *Framework) ensureService(name, namespace string, desiredEndpoints int) error {
 	return f.ensureServiceWithTimeout(name, namespace, desiredEndpoints, 120)
 }

 func (f *Framework) ensureServiceWithTimeout(name, namespace string, desiredEndpoints, timeout int) error {
 	backoff := wait.Backoff{
 		Duration: 6 * time.Second,
 		Factor:   1,
 		Steps:    timeout / 6,
 	}
 	var lastErr error
 	condFunc := func() (bool, error) {
 		ep, err := f.clientset.CoreV1().Endpoints(namespace).Get(f.Context, name, metav1.GetOptions{})
 		if err != nil {
 			lastErr = err
 			log.Errorw("failed to list endpoints",
 				zap.String("service", name),
 				zap.Error(err),
 			)
 			return false, nil
 		}
 		count := 0
 		for _, ss := range ep.Subsets {
 			count += len(ss.Addresses)
 		}
 		if count == desiredEndpoints {
 			return true, nil
 		}
 		log.Infow("endpoints count mismatch",
 			zap.String("service", name),
 			zap.Any("ep", ep),
 			zap.Int("expected", desiredEndpoints),
 			zap.Int("actual", count),
 		)
 		lastErr = fmt.Errorf("expected endpoints: %d but seen %d", desiredEndpoints, count)
 		return false, nil
 	}

 	err := wait.ExponentialBackoff(backoff, condFunc)
 	if err != nil {
 		return lastErr
 	}
 	return nil
 }

 func (f *Framework) GetServiceEndpoints(nn types.NamespacedName) ([]string, error) {
 	ep, err := f.clientset.CoreV1().Endpoints(cmp.Or(nn.Namespace, _namespace)).Get(f.Context, nn.Name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
 	var endpoints []string
 	for _, ss := range ep.Subsets {
 		for _, addr := range ss.Addresses {
 			endpoints = append(endpoints, addr.IP)
 		}
 	}
 	return endpoints, nil
 }

 //nolint:unused
 func (f *Framework) deletePods(selector string) {
 	podList, err := f.clientset.CoreV1().Pods(_namespace).List(f.Context, metav1.ListOptions{
 		LabelSelector: selector,
 	})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pods")
 	for _, pod := range podList.Items {
 		_ = f.clientset.CoreV1().
 			Pods(_namespace).
 			Delete(f.Context, pod.Name, metav1.DeleteOptions{GracePeriodSeconds: ptr.To(int64(30))})
 	}
 }

 func (f *Framework) CreateNamespaceWithTestService(name string) {
 	_, err := f.clientset.CoreV1().
 		Namespaces().
 		Create(f.Context, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name}}, metav1.CreateOptions{})
 	if err != nil && !errors.IsAlreadyExists(err) {
 		f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "create namespace")
 		return
 	}

 	_, err = f.clientset.CoreV1().Services(name).Create(f.Context, &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test",
 			Namespace: name,
 		},
 		Spec: corev1.ServiceSpec{
 			Ports: []corev1.ServicePort{
 				{
 					Name:     "http",
 					Port:     80,
 					Protocol: corev1.ProtocolTCP,
 				},
 			},
 			Selector: map[string]string{
 				"app": "httpbin",
 			},
 			Type: corev1.ServiceTypeClusterIP,
 		},
 	}, metav1.CreateOptions{})
 	if err != nil && !errors.IsAlreadyExists(err) {
 		f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "create service")
 	}
 }

 func (f *Framework) DeleteNamespace(name string) {
 	err := f.clientset.CoreV1().Namespaces().Delete(f.Context, name, metav1.DeleteOptions{})
 	if err == nil || errors.IsNotFound(err) {
 		return
 	}
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "delete namespace")
 }

 func (f *Framework) Scale(name string, replicas int32) {
 	scale, err := f.clientset.AppsV1().Deployments(_namespace).GetScale(context.Background(), name, metav1.GetOptions{})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("get deployment %s scale failed", name))
 	if scale.Spec.Replicas == replicas {
 		return
 	}
 	scale.Spec.Replicas = replicas
 	_, err = f.clientset.AppsV1().
 		Deployments(_namespace).
 		UpdateScale(context.Background(), name, scale, metav1.UpdateOptions{})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("scale deployment %s to %v failed", name, replicas))

 	// FIXME: The service name and the deployment name may not be the same
 	err = f.ensureService(name, _namespace, int(replicas))
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(),
 		fmt.Sprintf("ensure service %s/%s has %v endpoints failed", _namespace, name, replicas))
 }

 func (f *Framework) GetPodIP(namespace, selector string) string {
 	pods := f.GetPods(namespace, selector)
 	f.GomegaT.Expect(pods).ShouldNot(BeEmpty())
 	return pods[0].Status.PodIP
 }

 func (f *Framework) GetPods(namespace, selector string) []corev1.Pod {
 	podList, err := f.clientset.CoreV1().Pods(cmp.Or(namespace, _namespace)).List(f.Context, metav1.ListOptions{
 		LabelSelector: selector,
 	})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred())
 	return podList.Items
 }

 //nolint:unused
 func (f *Framework) applySSLSecret(namespace, name string, cert, pkey, caCert []byte) {
 	kind := "Secret"
 	apiVersion := "v1"
 	secretType := corev1.SecretTypeTLS
 	secret := applycorev1.SecretApplyConfiguration{
 		TypeMetaApplyConfiguration: applymetav1.TypeMetaApplyConfiguration{
 			Kind:       &kind,
 			APIVersion: &apiVersion,
 		},
 		ObjectMetaApplyConfiguration: &applymetav1.ObjectMetaApplyConfiguration{
 			Name: &name,
 		},
 		Data: map[string][]byte{
 			"tls.crt": cert,
 			"tls.key": pkey,
 			"ca.crt":  caCert,
 		},
 		Type: &secretType,
 	}

 	cli, err := k8s.GetKubernetesClientE(f.GinkgoT)
 	Expect(err).ToNot(HaveOccurred())

 	_, err = cli.CoreV1().Secrets(namespace).Apply(context.TODO(), &secret, metav1.ApplyOptions{
 		FieldManager: "e2e",
 	})
 	Expect(err).ToNot(HaveOccurred(), "apply secret")
 }

 func WaitPodsAvailable(t testing.TestingT, kubeOps *k8s.KubectlOptions, opts metav1.ListOptions) error {
 	condFunc := func() (bool, error) {
 		items, err := k8s.ListPodsE(t, kubeOps, opts)
 		if err != nil {
 			return false, err
 		}
 		if len(items) == 0 {
 			return false, nil
 		}
 		for _, item := range items {
 			foundPodReady := false
 			for _, cond := range item.Status.Conditions {
 				if cond.Type != corev1.PodReady {
 					continue
 				}
 				foundPodReady = true
 				if cond.Status != "True" {
 					return false, nil
 				}
 			}
 			if !foundPodReady {
 				return false, nil
 			}
 		}
 		return true, nil
 	}
 	return waitExponentialBackoff(condFunc)
 }

 func waitExponentialBackoff(condFunc func() (bool, error)) error {
 	backoff := wait.Backoff{
 		Duration: 500 * time.Millisecond,
 		Factor:   2,
 		Steps:    8,
 	}
 	return wait.ExponentialBackoff(backoff, condFunc)
 }

 func (f *Framework) NewExpectResponse(httpBody any) *httpexpect.Response {
 	body, err := json.Marshal(httpBody)
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred())

 	return httpexpect.NewResponse(f.GinkgoT, &http.Response{
 		Header: http.Header{
 			"Content-Type": []string{"application/json"},
 		},
 		Body: io.NopCloser(bytes.NewBuffer(body)),
 	})
 }

 // ListPods query pods by label selector.
 func (f *Framework) ListPods(selector string) []corev1.Pod {
 	pods, err := f.clientset.CoreV1().Pods(_namespace).List(context.TODO(), metav1.ListOptions{
 		LabelSelector: selector,
 	})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pod: ", selector)
 	return pods.Items
 }

 func (f *Framework) ListRunningPods(selector string) []corev1.Pod {
 	pods, err := f.clientset.CoreV1().Pods(_namespace).List(context.TODO(), metav1.ListOptions{
 		LabelSelector: selector,
 	})
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pod: ", selector)
 	runningPods := make([]corev1.Pod, 0)
 	for _, p := range pods.Items {
 		if p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil {
 			runningPods = append(runningPods, p)
 		}
 	}
 	return runningPods
 }

 // ExecCommandInPod exec cmd in specify pod and return the output from stdout and stderr
 func (f *Framework) ExecCommandInPod(podName string, cmd ...string) (string, string) {
 	req := f.clientset.CoreV1().RESTClient().Post().
 		Resource("pods").
 		Name(podName).
 		Namespace(_namespace).SubResource("exec")
 	req.VersionedParams(
 		&corev1.PodExecOptions{
 			Command: cmd,
 			Stdin:   false,
 			Stdout:  true,
 			Stderr:  true,
 			TTY:     false,
 		},
 		scheme.ParameterCodec,
 	)

 	var stdout, stderr bytes.Buffer
 	exec, err := remotecommand.NewSPDYExecutor(f.restConfig, "POST", req.URL())
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "request kubernetes exec api")
 	_ = exec.StreamWithContext(context.TODO(), remotecommand.StreamOptions{
 		Stdin:  nil,
 		Stdout: &stdout,
 		Stderr: &stderr,
 	})
 	return strings.TrimSpace(stdout.String()), strings.TrimSpace(stderr.String())
 }

 func (f *Framework) GetPodLogs(name string, previous bool) string {
 	reader, err := f.clientset.CoreV1().
 		Pods(_namespace).
 		GetLogs(name, &corev1.PodLogOptions{Previous: previous}).
 		Stream(context.Background())
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "get logs")
 	defer func() {
 		_ = reader.Close()
 	}()

 	logs, err := io.ReadAll(reader)
 	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "read all logs")

 	return string(logs)
 }

 func (f *Framework) WaitControllerManagerLog(keyword string, sinceSeconds int64, timeout time.Duration) {
 	f.WaitPodsLog("control-plane=controller-manager", keyword, sinceSeconds, timeout)
 }

 func (f *Framework) WaitPodsLog(selector, keyword string, sinceSeconds int64, timeout time.Duration) {
 	pods := f.ListRunningPods(selector)
 	wg := sync.WaitGroup{}
 	for _, p := range pods {
 		wg.Add(1)
 		go func(p corev1.Pod) {
 			defer wg.Done()
 			opts := corev1.PodLogOptions{Follow: true}
 			if sinceSeconds > 0 {
 				opts.SinceSeconds = ptr.To(sinceSeconds)
 			} else {
 				opts.TailLines = ptr.To(int64(0))
 			}
 			logStream, err := f.clientset.CoreV1().Pods(p.Namespace).GetLogs(p.Name, &opts).Stream(context.Background())
 			f.GomegaT.Expect(err).Should(BeNil())
 			scanner := bufio.NewScanner(logStream)
 			scanner.Split(bufio.ScanLines)
 			for scanner.Scan() {
 				line := scanner.Text()
 				if strings.Contains(line, keyword) {
 					return
 				}
 			}
 		}(p)
 	}
 	c := make(chan struct{})
 	go func() {
 		defer close(c)
 		wg.Wait()
 	}()
 	select {
 	case <-c:
 		return
 	case <-time.After(timeout):
 		f.GinkgoT.Error("wait log timeout")
 	}
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	package framework

	import (
	"bufio"
	"bytes"
	"cmp"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/api7/gopkg/pkg/log"
	"github.com/gavv/httpexpect/v2"
	"github.com/gruntwork-io/terratest/modules/k8s"
	"github.com/gruntwork-io/terratest/modules/testing"
	. "github.com/onsi/gomega" //nolint:staticcheck
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	applycorev1 "k8s.io/client-go/applyconfigurations/core/v1"
	applymetav1 "k8s.io/client-go/applyconfigurations/meta/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	"k8s.io/kubectl/pkg/scheme"
	"k8s.io/utils/ptr"
	)

	// buildRestConfig builds the rest.Config object from kubeconfig filepath and
	// context, if kubeconfig is missing, building from in-cluster configuration.
	func buildRestConfig(context string) (*rest.Config, error) {

	// Config loading rules:
	// 1. kubeconfig if it not empty string
	// 2. Config(s) in KUBECONFIG environment variable
	// 3. In cluster config if running in-cluster
	// 4. Use $HOME/.kube/config
	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
	loadingRules.DefaultClientConfig = &clientcmd.DefaultClientConfig
	configOverrides := &clientcmd.ConfigOverrides{
	ClusterDefaults: clientcmd.ClusterDefaults,
	CurrentContext: context,
	}

	clientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
	return clientConfig.ClientConfig()
	}

	func (f *Framework) ensureService(name, namespace string, desiredEndpoints int) error {
	return f.ensureServiceWithTimeout(name, namespace, desiredEndpoints, 120)
	}

	func (f *Framework) ensureServiceWithTimeout(name, namespace string, desiredEndpoints, timeout int) error {
	backoff := wait.Backoff{
	Duration: 6 * time.Second,
	Factor: 1,
	Steps: timeout / 6,
	}
	var lastErr error
	condFunc := func() (bool, error) {
	ep, err := f.clientset.CoreV1().Endpoints(namespace).Get(f.Context, name, metav1.GetOptions{})
	if err != nil {
	lastErr = err
	log.Errorw("failed to list endpoints",
	zap.String("service", name),
	zap.Error(err),
	)
	return false, nil
	}
	count := 0
	for _, ss := range ep.Subsets {
	count += len(ss.Addresses)
	}
	if count == desiredEndpoints {
	return true, nil
	}
	log.Infow("endpoints count mismatch",
	zap.String("service", name),
	zap.Any("ep", ep),
	zap.Int("expected", desiredEndpoints),
	zap.Int("actual", count),
	)
	lastErr = fmt.Errorf("expected endpoints: %d but seen %d", desiredEndpoints, count)
	return false, nil
	}

	err := wait.ExponentialBackoff(backoff, condFunc)
	if err != nil {
	return lastErr
	}
	return nil
	}

	func (f *Framework) GetServiceEndpoints(nn types.NamespacedName) ([]string, error) {
	ep, err := f.clientset.CoreV1().Endpoints(cmp.Or(nn.Namespace, _namespace)).Get(f.Context, nn.Name, metav1.GetOptions{})
	if err != nil {
	return nil, err
	}
	var endpoints []string
	for _, ss := range ep.Subsets {
	for _, addr := range ss.Addresses {
	endpoints = append(endpoints, addr.IP)
	}
	}
	return endpoints, nil
	}

	//nolint:unused
	func (f *Framework) deletePods(selector string) {
	podList, err := f.clientset.CoreV1().Pods(_namespace).List(f.Context, metav1.ListOptions{
	LabelSelector: selector,
	})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pods")
	for _, pod := range podList.Items {
	_ = f.clientset.CoreV1().
	Pods(_namespace).
	Delete(f.Context, pod.Name, metav1.DeleteOptions{GracePeriodSeconds: ptr.To(int64(30))})
	}
	}

	func (f *Framework) CreateNamespaceWithTestService(name string) {
	_, err := f.clientset.CoreV1().
	Namespaces().
	Create(f.Context, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name}}, metav1.CreateOptions{})
	if err != nil && !errors.IsAlreadyExists(err) {
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "create namespace")
	return
	}

	_, err = f.clientset.CoreV1().Services(name).Create(f.Context, &corev1.Service{
	ObjectMeta: metav1.ObjectMeta{
	Name: "test",
	Namespace: name,
	},
	Spec: corev1.ServiceSpec{
	Ports: []corev1.ServicePort{
	{
	Name: "http",
	Port: 80,
	Protocol: corev1.ProtocolTCP,
	},
	},
	Selector: map[string]string{
	"app": "httpbin",
	},
	Type: corev1.ServiceTypeClusterIP,
	},
	}, metav1.CreateOptions{})
	if err != nil && !errors.IsAlreadyExists(err) {
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "create service")
	}
	}

	func (f *Framework) DeleteNamespace(name string) {
	err := f.clientset.CoreV1().Namespaces().Delete(f.Context, name, metav1.DeleteOptions{})
	if err == nil \|\| errors.IsNotFound(err) {
	return
	}
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "delete namespace")
	}

	func (f *Framework) Scale(name string, replicas int32) {
	scale, err := f.clientset.AppsV1().Deployments(_namespace).GetScale(context.Background(), name, metav1.GetOptions{})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("get deployment %s scale failed", name))
	if scale.Spec.Replicas == replicas {
	return
	}
	scale.Spec.Replicas = replicas
	_, err = f.clientset.AppsV1().
	Deployments(_namespace).
	UpdateScale(context.Background(), name, scale, metav1.UpdateOptions{})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("scale deployment %s to %v failed", name, replicas))

	// FIXME: The service name and the deployment name may not be the same
	err = f.ensureService(name, _namespace, int(replicas))
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(),
	fmt.Sprintf("ensure service %s/%s has %v endpoints failed", _namespace, name, replicas))
	}

	func (f *Framework) GetPodIP(namespace, selector string) string {
	pods := f.GetPods(namespace, selector)
	f.GomegaT.Expect(pods).ShouldNot(BeEmpty())
	return pods[0].Status.PodIP
	}

	func (f *Framework) GetPods(namespace, selector string) []corev1.Pod {
	podList, err := f.clientset.CoreV1().Pods(cmp.Or(namespace, _namespace)).List(f.Context, metav1.ListOptions{
	LabelSelector: selector,
	})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred())
	return podList.Items
	}

	//nolint:unused
	func (f *Framework) applySSLSecret(namespace, name string, cert, pkey, caCert []byte) {
	kind := "Secret"
	apiVersion := "v1"
	secretType := corev1.SecretTypeTLS
	secret := applycorev1.SecretApplyConfiguration{
	TypeMetaApplyConfiguration: applymetav1.TypeMetaApplyConfiguration{
	Kind: &kind,
	APIVersion: &apiVersion,
	},
	ObjectMetaApplyConfiguration: &applymetav1.ObjectMetaApplyConfiguration{
	Name: &name,
	},
	Data: map[string][]byte{
	"tls.crt": cert,
	"tls.key": pkey,
	"ca.crt": caCert,
	},
	Type: &secretType,
	}

	cli, err := k8s.GetKubernetesClientE(f.GinkgoT)
	Expect(err).ToNot(HaveOccurred())

	_, err = cli.CoreV1().Secrets(namespace).Apply(context.TODO(), &secret, metav1.ApplyOptions{
	FieldManager: "e2e",
	})
	Expect(err).ToNot(HaveOccurred(), "apply secret")
	}

	func WaitPodsAvailable(t testing.TestingT, kubeOps *k8s.KubectlOptions, opts metav1.ListOptions) error {
	condFunc := func() (bool, error) {
	items, err := k8s.ListPodsE(t, kubeOps, opts)
	if err != nil {
	return false, err
	}
	if len(items) == 0 {
	return false, nil
	}
	for _, item := range items {
	foundPodReady := false
	for _, cond := range item.Status.Conditions {
	if cond.Type != corev1.PodReady {
	continue
	}
	foundPodReady = true
	if cond.Status != "True" {
	return false, nil
	}
	}
	if !foundPodReady {
	return false, nil
	}
	}
	return true, nil
	}
	return waitExponentialBackoff(condFunc)
	}

	func waitExponentialBackoff(condFunc func() (bool, error)) error {
	backoff := wait.Backoff{
	Duration: 500 * time.Millisecond,
	Factor: 2,
	Steps: 8,
	}
	return wait.ExponentialBackoff(backoff, condFunc)
	}

	func (f Framework) NewExpectResponse(httpBody any) httpexpect.Response {
	body, err := json.Marshal(httpBody)
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred())

	return httpexpect.NewResponse(f.GinkgoT, &http.Response{
	Header: http.Header{
	"Content-Type": []string{"application/json"},
	},
	Body: io.NopCloser(bytes.NewBuffer(body)),
	})
	}

	// ListPods query pods by label selector.
	func (f *Framework) ListPods(selector string) []corev1.Pod {
	pods, err := f.clientset.CoreV1().Pods(_namespace).List(context.TODO(), metav1.ListOptions{
	LabelSelector: selector,
	})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pod: ", selector)
	return pods.Items
	}

	func (f *Framework) ListRunningPods(selector string) []corev1.Pod {
	pods, err := f.clientset.CoreV1().Pods(_namespace).List(context.TODO(), metav1.ListOptions{
	LabelSelector: selector,
	})
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "list pod: ", selector)
	runningPods := make([]corev1.Pod, 0)
	for _, p := range pods.Items {
	if p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil {
	runningPods = append(runningPods, p)
	}
	}
	return runningPods
	}

	// ExecCommandInPod exec cmd in specify pod and return the output from stdout and stderr
	func (f *Framework) ExecCommandInPod(podName string, cmd ...string) (string, string) {
	req := f.clientset.CoreV1().RESTClient().Post().
	Resource("pods").
	Name(podName).
	Namespace(_namespace).SubResource("exec")
	req.VersionedParams(
	&corev1.PodExecOptions{
	Command: cmd,
	Stdin: false,
	Stdout: true,
	Stderr: true,
	TTY: false,
	},
	scheme.ParameterCodec,
	)

	var stdout, stderr bytes.Buffer
	exec, err := remotecommand.NewSPDYExecutor(f.restConfig, "POST", req.URL())
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "request kubernetes exec api")
	_ = exec.StreamWithContext(context.TODO(), remotecommand.StreamOptions{
	Stdin: nil,
	Stdout: &stdout,
	Stderr: &stderr,
	})
	return strings.TrimSpace(stdout.String()), strings.TrimSpace(stderr.String())
	}

	func (f *Framework) GetPodLogs(name string, previous bool) string {
	reader, err := f.clientset.CoreV1().
	Pods(_namespace).
	GetLogs(name, &corev1.PodLogOptions{Previous: previous}).
	Stream(context.Background())
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "get logs")
	defer func() {
	_ = reader.Close()
	}()

	logs, err := io.ReadAll(reader)
	f.GomegaT.Expect(err).ShouldNot(HaveOccurred(), "read all logs")

	return string(logs)
	}

	func (f *Framework) WaitControllerManagerLog(keyword string, sinceSeconds int64, timeout time.Duration) {
	f.WaitPodsLog("control-plane=controller-manager", keyword, sinceSeconds, timeout)
	}

	func (f *Framework) WaitPodsLog(selector, keyword string, sinceSeconds int64, timeout time.Duration) {
	pods := f.ListRunningPods(selector)
	wg := sync.WaitGroup{}
	for _, p := range pods {
	wg.Add(1)
	go func(p corev1.Pod) {
	defer wg.Done()
	opts := corev1.PodLogOptions{Follow: true}
	if sinceSeconds > 0 {
	opts.SinceSeconds = ptr.To(sinceSeconds)
	} else {
	opts.TailLines = ptr.To(int64(0))
	}
	logStream, err := f.clientset.CoreV1().Pods(p.Namespace).GetLogs(p.Name, &opts).Stream(context.Background())
	f.GomegaT.Expect(err).Should(BeNil())
	scanner := bufio.NewScanner(logStream)
	scanner.Split(bufio.ScanLines)
	for scanner.Scan() {
	line := scanner.Text()
	if strings.Contains(line, keyword) {
	return
	}
	}
	}(p)
	}
	c := make(chan struct{})
	go func() {
	defer close(c)
	wg.Wait()
	}()
	select {
	case <-c:
	return
	case <-time.After(timeout):
	f.GinkgoT.Error("wait log timeout")
	}
	}