package e2e

import (
	"context"
	"time"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	ep "github.com/tidbcloud/aws-shared-provider/pkg/utils/endpoint"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/kubeapi"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	apitypes "k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// var _ = Describe("cluster test internal", func() {
// 	interval := 10 * time.Second
// 	ctx := context.Background()
//
// 	It("cluster test internal", func() {
// 		By("Create a cluster and wait for ready")
// 		cluster := newCommonCluster()
// 		Expect(infraCli.Create(ctx, cluster)).Should(Succeed(), clusterStatus.KeyspaceToken)
// 		waitClusterReady(ctx, cluster, 5*time.Minute, interval)
//
// 		By("test TiDB Pod cannot access the kube-apiserver")
// 		// test by run "nc -z -w 5 -v kubernetes.default.svc 443"
// 		kubectlBuilder := NewKubectlCommand(
// 			eksKubeconfig,
// 			eksKubecontext,
// 			tcKey.Namespace,
// 			"exec",
// 			tcKey.Name+"-tidb-0",
// 			"-c",
// 			"tidb",
// 			"--",
// 			"nc",
// 			"-z",
// 			"-w",
// 			"5",
// 			"-v",
// 			"kubernetes.default.svc",
// 			"443",
// 		).WithStdinData("")
// 		stdout, stderr, err := kubectlBuilder.ExecWithFullOutput()
// 		Expect(err).ShouldNot(Succeed(), "stdout: %s, stderr: %s", stdout, stderr)
//
// 		By("delete cluster and check related resources in eks deleted")
// 		Expect(infraCli.Delete(ctx, cluster)).Should(Succeed(), clusterStatus.KeyspaceToken)
// 		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
// 	})
// })

var _ = Describe("tidb pods test health check", func() {
	interval := 10 * time.Second
	ctx := context.Background()

	It("test unhealthy tidb pod", func() {
		By("Create a new serverless cluster in vip pool with two replicas")
		cluster := newCommonCluster()
		cluster.Spec.TiDB.PoolSelector = vipTidbPool
		cluster.Spec.TiDB.Replicas = 2
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("Check if manager assign two tidb pod")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		var podName string
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, vipTidbPool, func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			for _, endpoint := range t.Endpoints {
				podName, _ = ep.ParsePodFromEndpoint(endpoint)
				break
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("patch tidb image to a invalid image, thus tidb pod will be unhealthy")
		tidbPod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName, Namespace: vipTidbPool}}
		patchData := []byte(`{"spec":{"containers":[{"name":"tidb","image":"gcr.io/pingcap-public/tidb:e2e-invalid"}]}}`)
		Expect(eksClient.Patch(ctx, tidbPod, client.RawPatch(apitypes.StrategicMergePatchType, patchData))).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			Expect(eksClient.Get(ctx, client.ObjectKeyFromObject(tidbPod), tidbPod)).Should(
				Succeed(), cluster.Status.KeyspaceToken)
			Expect(tidbPod.GetDeletionTimestamp()).Should(BeNil())
			Expect(tidbPod.Status.Phase).Should(Equal(corev1.PodRunning))
			if kubeapi.IsPodReady(tidbPod) {
				return errors.Errorf("wait tidb pod %v unready", tidbPod.Name)
			}
			return nil
		}, 30*time.Second, 10*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("recheck the assign cache again, the unhealthy tidb pod should not be removed")
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, vipTidbPool, func(t *tokenInfo) error {
			for _, endpoint := range t.Endpoints {
				podName, _ := ep.ParsePodFromEndpoint(endpoint)
				if podName == tidbPod.Name {
					return nil
				}
			}
			return errors.Errorf("the unhealthy tidb pod %v should not be removed", tidbPod.Name)
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("all connects successfully, even if there has a tidb pod is unhealthy")
		for i := 0; i < 5; i++ {
			Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
				Should(Succeed(), cluster.Status.KeyspaceToken)
		}

		By("delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("test delete unhealthy tidb pod", func() {
		By("Create a new serverless cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("establish a connection from mysql")
		db, err := endpointConnectedDB(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
		Expect(err).Should(Succeed(), "failed to connect db", cluster.Status.KeyspaceToken)
		defer db.Close()

		By("Check if manager assign a tidb pod")
		var podName string
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			for _, endpoint := range t.Endpoints {
				podName, _ = ep.ParsePodFromEndpoint(endpoint)
				break
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("patch tidb image to a invalid image")
		tidbPod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName, Namespace: cluster.Status.TidbPool}}
		patchData := []byte(`{"spec":{"containers":[{"name":"tidb","image":"gcr.io/pingcap-public/tidb:invalid"}]}}`)
		Expect(eksClient.Patch(ctx, tidbPod, client.RawPatch(apitypes.StrategicMergePatchType, patchData))).
			Should(Succeed(), cluster.Status.KeyspaceToken)

		By("wait tidb pod to be terminating")
		// because it's unhealthy and the connection will be disconnected by inner_error
		Eventually(func() error {
			// In zero backend mode, reconnect will trigger reassign, so query will not be failed
			// if db.QueryRowContext(ctx, "select 1").Err() == nil {
			// 	return errors.Errorf("should fail to connect tidb pod %v", tidbPod.Name)
			// }
			err := eksClient.Get(ctx, client.ObjectKeyFromObject(tidbPod), tidbPod)
			if err != nil {
				if apierrors.IsNotFound(err) {
					return nil
				} else {
					Expect(err).Should(Succeed(), cluster.Status.KeyspaceToken)
				}
			}
			if tidbPod.GetDeletionTimestamp() != nil {
				return nil
			}
			return errors.Errorf("should terminating tidb pod %v", tidbPod.Name)
		}, 44*time.Second, 2*time.Second).Should(Succeed(), "debug invalid pod", cluster.Status.KeyspaceToken)

		By("delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})
})

var _ = Describe("cluster test api", func() {
	interval := 10 * time.Second
	ctx := context.Background()

	It("cluster test depend on api only", func() {
		By("Create a new serverless cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("Test infra cluster privatelink")
		cluster.Spec.SqlEndpoints.AwsPrivatelinkService = &infra.AwsPrivatelinkServiceSpec{Enabled: true}
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.AvailabilityZones).ShouldNot(BeEmpty())
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.ServiceName).ShouldNot(BeEmpty())
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.PrivateDnsName).Should(ContainSubstring("-privatelink."))

		By("Test infra cluster public sqlEndpoint")
		publicEndpoint := *cluster.Status.SqlEndpoints.Public
		cluster.Spec.SqlEndpoints.Public.Enabled = false
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Status.SqlEndpoints.Public.Host).Should(BeEmpty(), cluster.Status.KeyspaceToken)
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, publicEndpoint)).
			ShouldNot(Succeed(), cluster.Status.KeyspaceToken)
		cluster.Spec.SqlEndpoints.Public.Enabled = true
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Status.SqlEndpoints.Public.Host).ShouldNot(BeEmpty(), cluster.Status.KeyspaceToken)
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Test infra cluster privatelink ZR")
		cluster.Spec.EnableZoneRedundant = "yes"
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Status.ZoneRedundantEnabled).Should(Equal("yes"))
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.AvailabilityZones).ShouldNot(BeEmpty())
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.ServiceName).ShouldNot(BeEmpty())
		Expect(cluster.Status.SqlEndpoints.AwsPrivatelinkService.PrivateDnsName).Should(ContainSubstring("-privatelink-zr."))

		By("Test infra cluster AnnoKeyResumeEtcdToken")
		cluster.Annotations[types.AnnoKeyResumeEtcdToken] = time.Now().Format(time.RFC3339)
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			if cluster.Annotations[types.AnnoKeyResumeEtcdToken] != "" {
				return errors.Errorf("cluster %v waiting for AnnoKeyResumeEtcdToken to be removed", cluster.Name)
			}
			return nil
		}, 30*time.Second, interval).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Check if manager assign one tidb pod")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Test update tidb two replicas")
		cluster.Spec.TiDB.Replicas = 2
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)

		By("Check if manager assign two tidb pod")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Test update tidb scale-in to one replica")
		cluster.Spec.TiDB.Replicas = 1
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)

		By("Check if manager assign back to one tidb pod")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("wait assigned tidb pod to free")
		Eventually(func() error {
			return checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
				if t.Replicas != cluster.Spec.TiDB.Replicas {
					return errors.Errorf("cluster %v replicas not equal", cluster.Name)
				}
				if len(t.Endpoints) > 0 {
					logger.Infow("endpoints not empty", "cluster", cluster.Name, "endpoints", t.Endpoints)
					return errors.Errorf("cluster %v endpoints not empty", cluster.Name)
				}
				return nil
			})
		}, 10*time.Minute, time.Minute).Should(Succeed(), cluster.Status.KeyspaceToken)

		// By("Test update loadBalancerSourceRanges")
		// waitTiDBReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 10*time.Minute, interval)
		// err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// 	err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 	if err != nil {
		// 		return err
		// 	}
		// 	cluster.Spec.SqlEndpoints.Public.LoadBalancerSourceRanges = []string{"1.1.1.1/32"}
		// 	err = infraCli.Update(ctx, cluster)
		// 	return err
		// })
		// Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		// waitTiDBNotReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 10*time.Minute, interval)
		//
		// err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// 	err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 	if err != nil {
		// 		return err
		// 	}
		// 	cluster.Spec.SqlEndpoints.Public.LoadBalancerSourceRanges = []string{"0.0.0.0/0"}
		// 	err = infraCli.Update(ctx, cluster)
		// 	return err
		// })
		// Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		// waitTiDBReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 10*time.Minute, interval)

		// By("Test throttler")
		// err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// 	err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 	if err != nil {
		// 		return err
		// 	}
		// 	// temperately reduce this to test.
		// 	cluster.Annotations[controllercluster.UpdateBandwidthDurationAnnKey] = "30s"
		// 	err = infraCli.Update(ctx, cluster)
		// 	return err
		// })
		// Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		// lastInfo, err := getTrafficUsageInfo(cluster)
		// Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		// limitSpeed := 10 * 1024
		// // ref https://github.com/tidbcloud/aws-shared-provider/issues/149
		// // after limit 10KiB, unable to write a 1MiB record.
		// // we write a small record, but this should not be too small to avoid time
		// // consume at network RTT.
		// recoredSize := limitSpeed * 10
		// // write 3 minute to test the speed
		// writeTime := time.Minute * 3
		// speed, err := testWrite(db, writeTime, recoredSize)
		// Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		//
		// var addedIngress float64
		// var nowInfo *controllercluster.TrafficUsageInfo
		// Eventually(func() error {
		// 	err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 	if err != nil {
		// 		return err
		// 	}
		// 	nowInfo, err = getTrafficUsageInfo(cluster)
		// 	Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		// 	addedIngress = nowInfo.IngressUsage - lastInfo.IngressUsage
		// 	if addedIngress >= float64(recoredSize)*0.8 {
		// 		return nil
		// 	}
		//
		// 	return errWait
		// }, 5*time.Minute, interval).
		// 	Should(Succeed(), "write %v, but added ingress only %v, now: %+v", recoredSize, addedIngress, nowInfo)
		// logger.Infof("now bandwidth info %+v", *nowInfo)
		// Expect(nowInfo.EgressUsage > lastInfo.EgressUsage).Should(BeTrue(), "EgressUsage should be added")
		//
		// if speed <= float64(2*limitSpeed) {
		// 	logger.Warn("speed only %d/s skip test throttler", speed)
		// } else {
		// 	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// 		err := infraCli.Get(ctx, client.ObjectKey{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 		if err != nil {
		// 			return err
		// 		}
		// 		cluster.Annotations[types.AnnoKeyIngress] = humanize.IBytes(uint64(limitSpeed)) + "/s"
		// 		err = infraCli.Update(ctx, cluster)
		// 		return err
		// 	})
		// 	Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		//
		// 	// wait sometime to let it take affect.
		// 	// how long it will take affects ?
		// 	time.Sleep(time.Second * 10)
		// 	// test speed by writing records
		// 	writeTime := time.Minute * 3
		// 	speed, err = testWrite(db, writeTime, recoredSize)
		// 	Expect(err).Should(Succeed(), clusterStatus.KeyspaceToken)
		//
		// 	accept := speed >= 0.5*float64(limitSpeed) && speed <= 1.5*float64(limitSpeed)
		// 	logger.Infof("speed %s/s after limit %s/s",
		// 		humanize.IBytes(uint64(speed)),
		// 		humanize.IBytes(uint64(limitSpeed)),
		// 	)
		// 	Expect(accept).Should(
		// 		BeTrue(),
		// 		"limit %s/s, but actual speed %s/s not in acceptable result range",
		// 		humanize.IBytes(uint64(limitSpeed)),
		// 		humanize.IBytes(uint64(speed)),
		// 	)
		// }

		By("delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("cluster test ensure resource group", func() {
		By("Create a new serverless cluster and set resource group")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "root", "rg1", "")).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != cluster.Status.TidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, cluster.Status.TidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Set another resource group with pool")
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "root", "rg2", vipTidbPool)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != vipTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, vipTidbPool)
			}
			if _, ok := cluster.Status.ResourceGroups["rg1"]; ok {
				return errors.Errorf("cluster %v waiting for rg1 be deleted", cluster.Name)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("another user use same resource group")
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "whl", "rg2", vipTidbPool)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgNameWhl := cluster.Spec.TiDB.UserResourceGroup["whl"]
			rgNameRoot := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgNameWhl != rgNameRoot {
				return errors.Errorf("cluster %v should use same resource group(%v!=%v)", cluster.Name, rgNameWhl, rgNameRoot)
			}
			if rgPool := cluster.Status.ResourceGroups[rgNameWhl].Pool; rgPool != vipTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgNameWhl, rgPool, vipTidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Delete whl resource group")
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "whl", "", "")).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgNameWhl := cluster.Spec.TiDB.UserResourceGroup["whl"]
			if rgNameWhl != "" {
				return errors.Errorf("cluster %v waiting for %v in user to be deleted", cluster.Name, rgNameWhl)
			}
			rgNameRoot := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgNameRoot == "" {
				return errors.Errorf("cluster %v root resource group should not be deleted", cluster.Name)
			}
			if rgPool := cluster.Status.ResourceGroups[rgNameRoot].Pool; rgPool != vipTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgNameRoot, rgPool, vipTidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Delete root resource group")
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "root", "", "")).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgName != "" {
				return errors.Errorf("cluster %v waiting for %v in user to be deleted", cluster.Name, rgName)
			}
			if _, ok := cluster.Status.ResourceGroups["rg2"]; ok {
				return errors.Errorf("cluster %v waiting for rg2 be deleted", cluster.Name)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("cluster test resource group pool change", func() {
		By("Create a new serverless cluster and set 2 resource groups")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "root", "rg1", "")).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "whl", "rg2", vipTidbPool)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		oldTidbPool := cluster.Status.TidbPool
		Expect(oldTidbPool).ShouldNot(BeEmpty())
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != cluster.Status.TidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, cluster.Status.TidbPool)
			}
			rgName = cluster.Spec.TiDB.UserResourceGroup["whl"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != vipTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, vipTidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Change rg2 to old tidb pool")
		Expect(setResourceGroupByToken(cluster.Status.KeyspaceToken, "whl", "rg2", oldTidbPool)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != cluster.Status.TidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, cluster.Status.TidbPool)
			}
			rgName = cluster.Spec.TiDB.UserResourceGroup["whl"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != oldTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, oldTidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Establish a connection from mysql")
		db, err := endpointConnectedDB(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
		Expect(err).Should(Succeed(), "failed to connect db", cluster.Status.KeyspaceToken)
		defer db.Close()

		By("Change cluster to vip pool")
		waitClusterUpdated(ctx, func() *infra.Cluster {
			cluster.Spec.TiDB.PoolSelector = vipTidbPool
			return cluster
		}, 30*time.Second, 2*time.Second)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Annotations).Should(HaveKey(types.AnnoKeyTidbPoolChanging), cluster.Status.KeyspaceToken)
		Expect(cluster.Status.TidbPool).Should(Equal(vipTidbPool), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				return err
			}
			rgName := cluster.Spec.TiDB.UserResourceGroup["root"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != cluster.Status.TidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, cluster.Status.TidbPool)
			}
			rgName = cluster.Spec.TiDB.UserResourceGroup["whl"]
			if rgPool := cluster.Status.ResourceGroups[rgName].Pool; rgPool != oldTidbPool {
				return errors.Errorf("cluster %v waiting for %v in status(%v!=%v)", cluster.Name, rgName, rgPool, oldTidbPool)
			}
			return nil
		}, 30*time.Second, 2*time.Second).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("cluster test tidb pool change", func() {
		By("Create a new serverless cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)
		oldTidbPool := cluster.Status.TidbPool
		Expect(oldTidbPool).ShouldNot(BeEmpty())

		By("Establish a connection from mysql")
		db, err := endpointConnectedDB(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
		Expect(err).Should(Succeed(), "failed to connect db", cluster.Status.KeyspaceToken)
		defer db.Close()

		By("Change cluster to vip pool")
		waitClusterUpdated(ctx, func() *infra.Cluster {
			cluster.Spec.TiDB.PoolSelector = vipTidbPool
			return cluster
		}, 30*time.Second, 2*time.Second)
		waitClusterSynced(ctx, cluster, 1*time.Minute, interval)
		Expect(cluster.Annotations).Should(HaveKey(types.AnnoKeyTidbPoolChanging), cluster.Status.KeyspaceToken)
		Expect(cluster.Status.TidbPool).Should(Equal(vipTidbPool), cluster.Status.KeyspaceToken)

		By("Check pods in old tidb pool")
		_ = checkTokenInfoFromManager(cluster.Status.KeyspaceToken, oldTidbPool, func(t *tokenInfo) error {
			Expect(len(t.Endpoints)).ShouldNot(Equal(0),
				"tidb pods in old pool should not empty", cluster.Status.KeyspaceToken)
			for _, endpoint := range t.Endpoints {
				podName, _ := ep.ParsePodFromEndpoint(endpoint)
				pod := &corev1.Pod{}
				Expect(eksClient.Get(ctx, client.ObjectKey{Namespace: oldTidbPool, Name: podName}, pod)).
					Should(Succeed(), cluster.Status.KeyspaceToken)
				// Expect(pod.Annotations).Should(HaveKey(types.AnnoTidbStatusRunSlow),
				// 	"tidb pods in old pool should have AnnoTidbStatusRunSlow", cluster.Status.KeyspaceToken)
			}
			return nil
		})

		By("Check the connection when tidb pool is changing")
		Expect(db.QueryRowContext(ctx, "select 1").Err()).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Wait cluster pool changed")
		waitClusterPoolChanged(ctx, cluster, 2*time.Minute, interval)

		By("Check pods in new tidb pool")
		_ = checkTokenInfoFromManager(cluster.Status.KeyspaceToken, vipTidbPool, func(t *tokenInfo) error {
			Expect(len(t.Endpoints)).ShouldNot(Equal(0),
				"tidb pods in new pool should not empty", cluster.Status.KeyspaceToken)
			for _, endpoint := range t.Endpoints {
				podName, _ := ep.ParsePodFromEndpoint(endpoint)
				pod := &corev1.Pod{}
				Expect(eksClient.Get(ctx, client.ObjectKey{Namespace: vipTidbPool, Name: podName}, pod)).
					Should(Succeed(), cluster.Status.KeyspaceToken)
				// Expect(pod.Annotations).ShouldNot(HaveKey(types.AnnoTidbStatusRunSlow),
				// 	"tidb pods in new pool should not have AnnoTidbStatusRunSlow", cluster.Status.KeyspaceToken)
			}
			return nil
		})

		By("Delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("cluster restore status test", func() {
		By("Create a new serverless cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("Check if manager assign one tidb pod")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			Should(Succeed(), cluster.Status.KeyspaceToken)
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, "", func(t *tokenInfo) error {
			if t.Replicas != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal", cluster.Name)
			}
			if len(t.Endpoints) != cluster.Spec.TiDB.Replicas {
				return errors.Errorf("cluster %v replicas not equal endpoints", cluster.Name)
			}
			return nil
		})).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("Restore cluster")
		waitClusterUpdated(ctx, func() *infra.Cluster {
			cluster.Annotations[types.AnnoKeyKeyspaceRestore] = types.AnnoKeyspaceRestoreStart
			return cluster
		}, 30*time.Second, 2*time.Second)
		time.Sleep(3 * time.Second)
		Eventually(func() error {
			err := infraClusterCli.Get(ctx, cluster)
			if err != nil {
				logger.Errorw("error inspecting cluster status", "cluster", cluster.Name, "error", err.Error())
				return err
			}
			if cluster.Annotations[types.AnnoKeyKeyspaceRestore] != types.AnnoKeyspaceRestartTiDB {
				return errWait
			}
			return nil
		}, 1*time.Minute, interval).Should(
			Succeed(),
			"Cluster should be synced spec: %v, status: %v", cluster.Spec, cluster.Status,
			cluster.Status.KeyspaceToken,
		)

		By("Check if manager stop the tidb-server process")
		Expect(endpointConnected(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)).
			ShouldNot(Succeed(), cluster.Status.KeyspaceToken)

		By("delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

	It("cluster test RU limit mode", func() {
		By("Create a new serverless cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("check mode is standard")
		Expect(cluster.Status.RuLimitMode).Should(Equal("standard"))

		By("change ru limit mode to throttled")
		cluster.Spec.RuLimitMode = throttledMode
		Expect(infraClusterCli.Update(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		Eventually(func() error {
			if err := infraClusterCli.Get(ctx, cluster); err != nil {
				return err
			}
			if cluster.Status.RuLimitMode != throttledMode {
				return errors.Errorf("cluster %s ru limit mode is not throttled", cluster.Name)
			}
			return nil
		}).WithTimeout(5*time.Minute).WithPolling(interval).Should(Succeed(), cluster.Status.KeyspaceToken)

		By("delete cluster and wait it to be deleted")
		Expect(infraClusterCli.Delete(ctx, cluster)).Should(Succeed(), cluster.Status.KeyspaceToken)
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})
})

// skip the audit log test because the kernel audit log config are now controlled by tidb global system variable, which are set in controller plane.
// var _ = Describe("cluster audit log test", func() {
// 	interval := 10 * time.Second
// 	ctx := context.Background()

// 	It("cluster with audit log enabled can see logs in stdout", func() {
// 		By("Create a new serverless cluster1 with audit log enabled")
// 		cluster1 := newCommonCluster()
// 		cluster1.Spec.TiDB.Audit = true
// 		Expect(infraClusterCli.Create(ctx, cluster1)).Should(Succeed())
// 		waitClusterReady(ctx, cluster1, 5*time.Minute, interval)

// 		logger.Infow("cluster1 created", "name", cluster1.Name)

// 		By("establish connection to cluster1 from mysql")
// 		db, err := endpointConnectedDBWithCloudAdmin(ctx, cluster1.Status.KeyspaceToken, *cluster1.Status.SqlEndpoints.Public)
// 		Expect(err).Should(Succeed(), "failed to connect db1", cluster1.Status.KeyspaceToken)
// 		_, err = db.Exec("SELECT audit_log_create_filter('all', '{}');")
// 		Expect(err).Should(Succeed(), "failed to create filter", cluster1.Status.KeyspaceToken)
// 		_, err = db.Exec("SELECT audit_log_create_rule('%@%', 'all');")
// 		Expect(err).Should(Succeed(), "failed to create filter rule", cluster1.Status.KeyspaceToken)
// 		_, err = db.Exec("SHOW DATABASES;")
// 		Expect(err).Should(Succeed(), "failed to exec sql", cluster1.Status.KeyspaceToken)
// 		defer db.Close()

// 		var podName1 string
// 		Expect(checkTokenInfoFromManager(cluster1.Status.KeyspaceToken, cluster1.Status.TidbPool, func(t *tokenInfo) error {
// 			for _, endpoint := range t.Endpoints {
// 				podName1, _ = ep.ParsePodFromEndpoint(endpoint)
// 				break
// 			}
// 			return nil
// 		})).Should(Succeed(), "failed to get pod name 1 from manager", cluster1.Status.KeyspaceToken)

// 		By("check with kubectl to see if audit log emitted when audit log enabled")
// 		kubectlBuilder := NewKubectlCommand(
// 			eksKubeconfig,
// 			eksKubecontext,
// 			cluster1.Status.TidbPool,
// 			"logs",
// 			podName1,
// 			"-c",
// 			"auditlog",
// 		).WithStdinData("")
// 		stdout, stderr, err := kubectlBuilder.ExecWithFullOutput()
// 		Expect(err).Should(Succeed(), "pod1 stdout: %s, stderr: %s", stdout, stderr, cluster1.Status.KeyspaceToken)
// 		Expect(strings.Contains(stdout, "QUERY")).Should(Equal(true), "pod1 stdout: %s, stderr: %s", stdout, stderr)

// 		By("edit cluster1 by disable audit log and expect to see audit log don't emit")
// 		logger.Infow("cluster1 spec", "audit log spec", cluster1.Spec.TiDB, cluster1.ResourceVersion)
// 		cluster1.Spec.TiDB.Audit = false
// 		Expect(infraClusterCli.Update(ctx, cluster1)).Should(Succeed(), cluster1.Status.KeyspaceToken)
// 		logger.Infow("cluster1 spec", "audit log spec", cluster1.Spec.TiDB, cluster1.ResourceVersion)
// 		// wait cache synced
// 		time.Sleep(time.Second * 10)
// 		waitClusterReady(ctx, cluster1, 5*time.Minute, interval)

// 		By("establish connection to cluster1 from mysql")
// 		db, err = endpointConnectedDB(ctx, cluster1.Status.KeyspaceToken, *cluster1.Status.SqlEndpoints.Public)
// 		Expect(err).Should(Succeed(), "failed to connect db1")
// 		_, err = db.Exec("SHOW DATABASES;")
// 		Expect(err).Should(Succeed(), "failed to exec sql", cluster1.Status.KeyspaceToken)
// 		defer db.Close()

// 		var podName3 string
// 		Expect(checkTokenInfoFromManager(cluster1.Status.KeyspaceToken, cluster1.Status.TidbPool, func(t *tokenInfo) error {
// 			for _, endpoint := range t.Endpoints {
// 				podName3, _ = ep.ParsePodFromEndpoint(endpoint)
// 				break
// 			}
// 			return nil
// 		})).Should(Succeed(), "failed to get pod name 1 from manager", cluster1.Status.KeyspaceToken)
// 		Expect(podName3).ShouldNot(Equal(podName1), "after edit audit log enabled, pod name should change")

// 		By("check with kubectl to see if audit log emitted when audit log disabled")
// 		kubectlBuilder = NewKubectlCommand(
// 			eksKubeconfig,
// 			eksKubecontext,
// 			cluster1.Status.TidbPool,
// 			"logs",
// 			podName3,
// 			"-c",
// 			"auditlog",
// 		).WithStdinData("")
// 		stdout, stderr, err = kubectlBuilder.ExecWithFullOutput()
// 		Expect(err).Should(Succeed(), "pod1 stdout: %s, stderr: %s", stdout, stderr, cluster1.Status.KeyspaceToken)
// 		Expect(strings.Contains(stdout, "QUERY")).Should(Equal(false), "pod1 stdout: %s, stderr: %s", stdout, stderr)

// 		By("delete cluster and wait it to be deleted")
// 		Expect(infraClusterCli.Delete(ctx, cluster1)).Should(Succeed(), cluster1.Status.KeyspaceToken)
// 		waitClusterDeleted(ctx, cluster1, 5*time.Minute, interval)
// 	})

// 	It("cluster with audit log disabled can't see logs in stdout", func() {
// 		By("Create a new serverless cluster2 with audit log disabled")
// 		cluster2 := newCommonCluster()
// 		Expect(infraClusterCli.Create(ctx, cluster2)).Should(Succeed())
// 		waitClusterReady(ctx, cluster2, 5*time.Minute, interval)

// 		logger.Infow("cluster2 created", "name", cluster2.Name)

// 		By("establish connection to cluster2 from mysql")
// 		db, err := endpointConnectedDBWithCloudAdmin(ctx, cluster2.Status.KeyspaceToken, *cluster2.Status.SqlEndpoints.Public)
// 		Expect(err).Should(Succeed(), "failed to connect db2", cluster2.Status.KeyspaceToken)
// 		_, err = db.Exec("SELECT audit_log_create_filter('all', '{}');")
// 		Expect(err).Should(Succeed(), "failed to create filter", cluster2.Status.KeyspaceToken)
// 		_, err = db.Exec("SELECT audit_log_create_rule('%@%', 'all');")
// 		Expect(err).Should(Succeed(), "failed to create filter rule", cluster2.Status.KeyspaceToken)
// 		_, err = db.Exec("SHOW DATABASES;")
// 		Expect(err).Should(Succeed(), "failed to exec sql", cluster2.Status.KeyspaceToken)
// 		defer db.Close()

// 		var podName2 string
// 		Expect(checkTokenInfoFromManager(cluster2.Status.KeyspaceToken, cluster2.Status.TidbPool, func(t *tokenInfo) error {
// 			for _, endpoint := range t.Endpoints {
// 				podName2, _ = ep.ParsePodFromEndpoint(endpoint)
// 				break
// 			}
// 			return nil
// 		})).Should(Succeed(), "failed to get pod name 2 from manager", cluster2.Status.KeyspaceToken)

// 		By("check with kubectl to see if audit log emitted when audit log disabled")
// 		kubectlBuilder := NewKubectlCommand(
// 			eksKubeconfig,
// 			eksKubecontext,
// 			cluster2.Status.TidbPool,
// 			"logs",
// 			podName2,
// 			"-c",
// 			"auditlog",
// 		).WithStdinData("")
// 		stdout, stderr, err := kubectlBuilder.ExecWithFullOutput()
// 		Expect(err).Should(Succeed(), "pod2 stdout: %s, stderr: %s", stdout, stderr, cluster2.Status.KeyspaceToken)
// 		Expect(strings.Contains(stdout, "QUERY")).Should(Equal(false), "pod2 stdout: %s, stderr: %s", stdout, stderr)

// 		By("edit cluster2 by enable audit log and expect to see audit log emit")
// 		cluster2.Spec.TiDB.Audit = true
// 		Expect(infraClusterCli.Update(ctx, cluster2)).Should(Succeed(), cluster2.Status.KeyspaceToken)
// 		// wait cache synced
// 		time.Sleep(time.Second * 10)
// 		waitClusterReady(ctx, cluster2, 5*time.Minute, interval)

// 		By("establish connection to cluster2 from mysql")
// 		db, err = endpointConnectedDB(ctx, cluster2.Status.KeyspaceToken, *cluster2.Status.SqlEndpoints.Public)
// 		Expect(err).Should(Succeed(), "failed to connect db2", cluster2.Status.KeyspaceToken)
// 		_, err = db.Exec("SHOW DATABASES;")
// 		Expect(err).Should(Succeed(), "failed to exec sql", cluster2.Status.KeyspaceToken)
// 		defer db.Close()

// 		var podName3 string
// 		Expect(checkTokenInfoFromManager(cluster2.Status.KeyspaceToken, cluster2.Status.TidbPool, func(t *tokenInfo) error {
// 			for _, endpoint := range t.Endpoints {
// 				podName3, _ = ep.ParsePodFromEndpoint(endpoint)
// 				break
// 			}
// 			return nil
// 		})).Should(Succeed(), "failed to get pod name 2 from manager", cluster2.Status.KeyspaceToken)
// 		Expect(podName3).ShouldNot(Equal(podName2), "after edit audit log enabled, pod name should change")

// 		By("check with kubectl to see if audit log emitted when audit log enabled")
// 		kubectlBuilder = NewKubectlCommand(
// 			eksKubeconfig,
// 			eksKubecontext,
// 			cluster2.Status.TidbPool,
// 			"logs",
// 			podName3,
// 			"-c",
// 			"auditlog",
// 		).WithStdinData("")
// 		stdout, stderr, err = kubectlBuilder.ExecWithFullOutput()
// 		Expect(err).Should(Succeed(), "pod2 stdout: %s, stderr: %s", stdout, stderr, cluster2.Status.KeyspaceToken)
// 		Expect(strings.Contains(stdout, "QUERY")).Should(Equal(true), "pod2 stdout: %s, stderr: %s", stdout, stderr)

// 		By("delete cluster and wait it to be deleted")
// 		Expect(infraClusterCli.Delete(ctx, cluster2)).Should(Succeed(), cluster2.Status.KeyspaceToken)
// 		waitClusterDeleted(ctx, cluster2, 5*time.Minute, interval)
// 	})
// })

var _ = Describe("zero backend mode test", func() {
	interval := 10 * time.Second
	ctx := context.Background()

	It("gateway can restore session in zero backend mode", func() {
		By("Create a cluster")
		cluster := newCommonCluster()
		Expect(infraClusterCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 5*time.Minute, interval)

		By("establish connection to cluster from mysql")
		db, err := endpointConnectedDB(ctx, cluster.Status.KeyspaceToken, *cluster.Status.SqlEndpoints.Public)
		Expect(err).Should(Succeed(), "failed to connect db", cluster.Status.KeyspaceToken)
		defer db.Close()

		By("update system variable")
		_, err = db.Exec("SET tidb_distsql_scan_concurrency = 10")
		Expect(err).Should(Succeed(), "failed to update system variable", cluster.Status.KeyspaceToken)
		row := db.QueryRow("SHOW VARIABLES LIKE ?", "tidb_distsql_scan_concurrency")
		var variableName string
		var concurrencyBefore int
		err = row.Scan(&variableName, &concurrencyBefore)
		Expect(err).Should(Succeed(), "failed to query system variable", cluster.Status.KeyspaceToken)
		Expect(concurrencyBefore).Should(Equal(10), "system variable value is not correct")

		var podsBefore []string
		Expect(checkTokenInfoFromManager(cluster.Status.KeyspaceToken, cluster.Status.TidbPool, func(t *tokenInfo) error {
			for _, endpoint := range t.Endpoints {
				podName, _ := ep.ParsePodFromEndpoint(endpoint)
				podsBefore = append(podsBefore, podName)
			}
			if len(podsBefore) == 0 {
				return errors.New("backend pods empty")
			}
			return nil
		})).Should(Succeed(), "failed to get pod name from manager", cluster.Status.KeyspaceToken)

		// max-idles-seconds is 60, reconnect when backend is shutting down
		time.Sleep(time.Second * 65)

		By("query again, expect to trigger backend reconnect")
		rowAfter := db.QueryRow("SHOW VARIABLES LIKE ?", "tidb_distsql_scan_concurrency")
		var concurrencyAfter int
		err = rowAfter.Scan(&variableName, &concurrencyAfter)
		Expect(err).Should(Succeed(), "failed to query system variable", cluster.Status.KeyspaceToken)
		Expect(concurrencyAfter).Should(Equal(concurrencyBefore), "system variable value should not change")

		// Occasionally, reconnect maybe assigned with the same pod if it's already released.
		// We just skip the test cases for now.
		// var podsAfter []string
		// Expect(checkTokenInfoFromManager(clusterStatus.KeyspaceToken, clusterStatus.TidbPool, func(t *tokenInfo) error {
		// 	for _, endpoint := range t.Endpoints {
		// 		podName, _ := utils.ParsePodFromEndpoint(endpoint)
		// 		podsAfter = append(podsAfter, podName)
		// 	}
		// 	if len(podsAfter) == 0 {
		// 		return errors.New("backend pods empty")
		// 	}
		// 	return nil
		// })).Should(Succeed(), "failed to get pod name from manager", clusterStatus.KeyspaceToken)
		//
		// Expect(len(podsBefore) != len(podsAfter) || podsBefore[0] != podsAfter[0]).Should(
		// 	Equal(true), "pods info should change when reconnect", clusterStatus.KeyspaceToken, podsBefore, podsAfter,
		// )

	})
})

// func getTrafficUsageInfo(cluster *cloudv1.Cluster) (*controllercluster.TrafficUsageInfo, error) {
// 	info := new(controllercluster.TrafficUsageInfo)
// 	data, ok := cluster.Annotations["bandwidth.tidbcloud.com/usage"]
// 	if !ok {
// 		return nil, errors.New("no bandwidth.tidbcloud.com/usage")
// 	}
//
// 	return info, json.Unmarshal([]byte(data), info)
// }

// func checkClusterStatusAfterRunning(cluster *cloudv1.Cluster) error {
// 	// check Metrics base url
// 	if cluster.Status.Metrics == nil {
// 		return errors.New("Metrics not ready")
// 	}
// 	str := cluster.Status.Metrics.BaseURL
// 	if !strings.HasPrefix(str, "http") {
// 		str = "https://" + str
// 	}
// 	u, err := url.Parse(str)
// 	if err != nil {
// 		return err
// 	}
// 	tcKey := clusterKeyToTCKey(client.ObjectKeyFromObject(cluster))
// 	expect := fmt.Sprintf("/proxy/namespace/%s/tidbmonitor/%s/prometheus", tcKey.Namespace, tcKey.Name)
// 	if u.Path != expect {
// 		return errors.Errorf("path is %q, but want %q", u.Path, expect)
// 	}
//
// 	// check proxy url
// 	if cluster.Status.Proxy == nil {
// 		return errors.New("Proxy status not ready")
// 	}
// 	str = cluster.Status.Proxy.Endpoint
// 	if !strings.HasPrefix(str, "http") {
// 		str = "https://" + str
// 	}
// 	u, err = url.Parse(str)
// 	if err != nil {
// 		return err
// 	}
// 	expect = fmt.Sprintf("/proxy/namespace/%s/cluster/%s", tcKey.Namespace, tcKey.Name)
// 	if u.Path != expect {
// 		return errors.Errorf("path is %q, but want %q", u.Path, expect)
// 	}
//
// 	return nil
// }

// func initRecordTable(ctx context.Context, db *sql.DB) error {
// 	s := "create database if not exists test"
// 	_, err := db.ExecContext(ctx, s)
// 	if err != nil {
// 		return err
// 	}
//
// 	s = "create table if not exists test.e2etest(id int auto_increment primary key, v LONGBLOB)"
// 	_, err = db.ExecContext(ctx, s)
// 	if err != nil {
// 		return err
// 	}
//
// 	return nil
// }

// writeRecord write a record with specify size.
// record will be written to the flowing table.
// test.e2etest(id int auto_increment primary key, v blob)
// func writeRecord(ctx context.Context, db *sql.DB, size int) error {
// 	data := make([]byte, size)
// 	_, err := db.ExecContext(ctx, "insert into test.e2etest(v) values(?)", data)
// 	if err != nil {
// 		return err
// 	}
//
// 	return nil
// }

// func testWrite(db *sql.DB, writeTime time.Duration, recoredSize int) (speed float64, err error) {
// 	logger.Infof("start writing %v time to test speed", writeTime)
// 	start := time.Now()
// 	var writeBytes int
// 	for {
// 		err := writeRecord(context.Background(), db, recoredSize)
// 		if err != nil {
// 			return 0, err
// 		}
//
// 		writeBytes += recoredSize
// 		if time.Since(start) > writeTime {
// 			break
// 		}
// 	}
// 	takes := time.Since(start)
// 	speed = float64(writeBytes) / takes.Seconds()
// 	logger.Infof("speed is %s/s", humanize.IBytes(uint64(speed)))
// 	return
// }