package e2e

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net/url"
	"reflect"
	"strings"
	"time"

	cmv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/pingcap/label"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	utilns "github.com/tidbcloud/shared-provider/pkg/utils/namespace"
	evictpod "github.com/tidbcloud/shared-provider/pkg/utils/pod"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var mockConfigMapKey = types.NamespacedName{Namespace: "tidb-admin", Name: "config-service-mock-cm"}

func clusterKeyToTCKey(key types.NamespacedName) types.NamespacedName {
	return types.NamespacedName{
		Namespace: utilns.NamespacePrefix + key.Namespace,
		Name:      "db" + key.Name,
	}
}

// checkConfig verify the config of tc as the configmap which mock-config-server uses.
func checkConfig(tc *v1alpha1.TidbCluster, cm *corev1.ConfigMap) {
	config := new(api.DescribeClusterConfigOutput)
	data := cm.Data["cluster-config-shared"]
	err := json.Unmarshal([]byte(data), config)
	Expect(err).Should(Succeed(), "failed to unmarshal %s", data)

	Expect(tc.Spec.Version).Should(Equal(config.Version))

	var tidbOutput *api.ComponentOutput
	for _, c := range config.Components {
		if c.Component == api.ComponentTiDB {
			tidbOutput = &c
			break
		}
	}
	Expect(tidbOutput).ShouldNot(BeNil(), "no tidb component in config")

	tcValue := tc.Spec.TiDB.Config.Get("max-server-connections").MustInt()
	cmValue := reflect.ValueOf(tidbOutput.Config.MP["max-server-connections"]).Int()
	Expect(tcValue).Should(Equal(cmValue))

	tcEnableSEM := tc.Spec.TiDB.Config.Get("security.enable-sem").Interface().(bool)
	Expect(tcEnableSEM).Should(BeTrue()) // enable-sem is always true since newly created cluster > 5.3.0
}

func getTidbAllPods(ctx context.Context, tcKey client.ObjectKey) []corev1.Pod {
	selector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
		MatchLabels: map[string]string{
			label.InstanceLabelKey: tcKey.Name,
		},
		MatchExpressions: []metav1.LabelSelectorRequirement{{
			Key:      label.ComponentLabelKey,
			Operator: metav1.LabelSelectorOpIn,
			Values:   evictpod.EvictComponents,
		}},
	})
	Expect(err).Should(Succeed())

	podList := &corev1.PodList{}
	err = eksClient.List(ctx, podList, &client.ListOptions{Namespace: tcKey.Namespace, LabelSelector: selector})
	Expect(err).Should(Succeed())
	Expect(len(podList.Items)).ShouldNot(BeZero())

	return podList.Items
}

func getTidbPod(ctx context.Context, tcKey client.ObjectKey) []corev1.Pod {
	selector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
		MatchLabels: map[string]string{
			label.InstanceLabelKey:  tcKey.Name,
			label.ComponentLabelKey: "tidb",
		},
	})
	Expect(err).Should(Succeed())

	podList := &corev1.PodList{}
	err = eksClient.List(ctx, podList, &client.ListOptions{Namespace: tcKey.Namespace, LabelSelector: selector})
	Expect(err).Should(Succeed())

	return podList.Items
}

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

	if eksKubeconfig == "" {
		logger.Info("do not specify eks to be test, skip testing internal case")
		return
	}
	It("cluster test internal", func() {
		By("Get the config map to specify config of cluster")
		cm := new(corev1.ConfigMap)
		err := eksClient.Get(ctx, mockConfigMapKey, cm)
		Expect(err).Should(Succeed())

		By("Create a cluster and wait for ready")
		cluster := newCommonCluster(awsNetwork)
		Expect(infraCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 15*time.Minute, interval)

		By("Check the cluster keeps Running") // no upgrade occur when enabling SEM
		Eventually(func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			if cluster.Status.Phase != cloudv1.ClusterPhaseRunning {
				return errors.Errorf("cluster %s is not running, current phase: %s", cluster.Name, cluster.Status.Phase)
			}
			return nil
		}, 3*time.Minute, interval).Should(Succeed())

		By("Check config as specified")
		tc := new(v1alpha1.TidbCluster)
		clusterKey := client.ObjectKeyFromObject(cluster)
		err = eksClient.Get(ctx, clusterKeyToTCKey(clusterKey), tc)
		Expect(err).Should(Succeed())
		checkConfig(tc, cm)

		By("Create another cluster and wait for ready")
		cluster2 := newCommonCluster(awsNetwork2)
		cluster2.Namespace = namespace2
		Expect(infraCli.Create(ctx, cluster2)).Should(Succeed())
		waitClusterReady(ctx, cluster2, 15*time.Minute, interval)

		By("Test TiDB Pod cannot access the Pods in other namespaces")
		tcKey := clusterKeyToTCKey(clusterKey)
		// run kubectl like:
		// k exec tcname-tidb-0 -n shared-tier-hjh -c tidb -- nc -z -w 5 -v tcname-tidb 4000
		// to test tidb pod in cluster can access itself but can not access tidb svc of cluster2(in another ns).
		kubectlBuilder := NewKubectlCommand(
			eksKubeconfig,
			eksKubecontext,
			tcKey.Namespace,
			"exec",
			tcKey.Name+"-tidb-0",
			"-c",
			"tidb",
			"--",
			"nc",
			"-z",
			"-w", // -w SEC	Timeout for connects and final net reads
			"5",
			"-v",
			tcKey.Name+"-tidb",
			"4000",
		).WithStdinData("")
		stdout, stderr, err := kubectlBuilder.ExecWithFullOutput()
		Expect(err).Should(Succeed(), "stdout: %s, stderr: %s", stdout, stderr)

		tcKey2 := clusterKeyToTCKey(client.ObjectKeyFromObject(cluster2))
		kubectlBuilder = NewKubectlCommand(
			eksKubeconfig,
			eksKubecontext,
			tcKey.Namespace,
			"exec",
			tcKey.Name+"-tidb-0",
			"-c",
			"tidb",
			"--",
			"nc",
			"-z",
			"-w",
			"5",
			"-v",
			fmt.Sprintf("%s.%s.svc", tcKey2.Name+"-tidb", tcKey2.Namespace),
			"4000",
		).WithStdinData("")
		stdout, stderr, err = kubectlBuilder.ExecWithFullOutput()
		Expect(err).ShouldNot(Succeed(), "stdout: %s, stderr: %s", stdout, stderr)

		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())
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
		err = checkClusterResoucesInEKSDeleted(ctx, eksClient, cluster)
		Expect(err).Should(Succeed())
		Expect(infraCli.Delete(ctx, cluster2)).Should(Succeed())
		waitClusterDeleted(ctx, cluster2, 5*time.Minute, interval)
		err = checkClusterResoucesInEKSDeleted(ctx, eksClient, cluster2)
		Expect(err).Should(Succeed())
	})

})

var _ = Describe("cluster test depend on api only", func() {
	interval := 10 * time.Second
	ctx := context.Background()
	It("handle SafeToEvictAnno on pod", func() {
		By("Create a new cluster")
		cluster := newCommonCluster(awsNetwork)
		Expect(infraCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 15*time.Minute, interval)

		By("default tidb pods should have SafeToEvictAnno")
		pods := getTidbAllPods(ctx, clusterKeyToTCKey(client.ObjectKeyFromObject(cluster)))
		for i := 0; i < len(pods); i++ {
			_, ok := pods[i].Annotations[evictpod.SafeToEvictAnno]
			Expect(ok).Should(BeTrue())
		}

		By("handleSafeToEvictAnno should remove SafeToEvictAnno")
		err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
			if err := infraCli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster); err != nil {
				return err
			}
			cluster.Annotations[utils.DetectActiveDurationAnnKey] = "10s"
			return infraCli.Update(ctx, cluster)
		})
		Expect(err).Should(Succeed())
		time.Sleep(30 * time.Second)
		pods = getTidbAllPods(ctx, clusterKeyToTCKey(client.ObjectKeyFromObject(cluster)))
		for i := 0; i < len(pods); i++ {
			_, ok := pods[i].Annotations[evictpod.SafeToEvictAnno]
			Expect(ok).Should(BeFalse())
		}

		By("handleSafeToEvictAnno should add SafeToEvictAnno")
		err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
			if err := infraCli.Get(ctx, client.ObjectKeyFromObject(cluster), cluster); err != nil {
				return err
			}
			now := metav1.Now()
			cluster.Status.LastActiveTimestamp = &now
			delete(cluster.Annotations, utils.DetectActiveDurationAnnKey)
			return infraCli.Update(ctx, cluster)
		})
		Expect(err).Should(Succeed())
		time.Sleep(10 * time.Second)
		pods = getTidbAllPods(ctx, clusterKeyToTCKey(client.ObjectKeyFromObject(cluster)))
		for i := 0; i < len(pods); i++ {
			_, ok := pods[i].Annotations[evictpod.SafeToEvictAnno]
			Expect(ok).Should(BeTrue())
		}

		By("should reAdd SafeToEvictAnno after delete a pod")
		pods = getTidbPod(ctx, clusterKeyToTCKey(client.ObjectKeyFromObject(cluster)))
		for i := 0; i < len(pods); i++ {
			err = eksClient.Delete(ctx, &pods[i], &client.DeleteOptions{GracePeriodSeconds: pointer.Int64Ptr(0)})
			Expect(err).Should(Succeed())
		}
		Eventually(func() error {
			pods = getTidbPod(ctx, clusterKeyToTCKey(client.ObjectKeyFromObject(cluster)))
			if len(pods) == 0 {
				return errors.New("no tidb pod started")
			}
			for i := 0; i < len(pods); i++ {
				_, ok := pods[i].Annotations[evictpod.SafeToEvictAnno]
				if !ok {
					return errors.New("tidb pods not have SafeToEvictAnno")
				}
			}
			return nil
		}, 2*time.Minute, interval).Should(Succeed())

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

	It("cluster test depend on api only", func() {
		By("Create a new cluster")
		cluster := newCommonCluster(awsNetwork)
		Expect(infraCli.Create(ctx, cluster)).Should(Succeed())
		waitClusterReady(ctx, cluster, 15*time.Minute, interval)
		checkClusterStatusAfterRunning(cluster)
		// although the cluster phase is running, but the Metrics status may still not set yet.
		Eventually(func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			return checkClusterStatusAfterRunning(cluster)
		}, 2*time.Minute, interval).Should(Succeed())

		By("Test update replicas")
		err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			cluster.Spec.Cluster.TiDB.Replicas++
			err = infraCli.Update(ctx, cluster)
			return err
		})
		Expect(err).Should(Succeed())
		waitClusterSynced(ctx, cluster, 15*time.Minute, interval)

		err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			cluster.Spec.Cluster.TiDB.Replicas--
			err = infraCli.Update(ctx, cluster)
			return err
		})
		Expect(err).Should(Succeed())
		waitClusterSynced(ctx, cluster, 15*time.Minute, interval)

		By("Test update loadBalancerSourceRanges")
		waitTiDBReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 15*time.Minute, interval)
		err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
			err := infraCli.Get(ctx, types.NamespacedName{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())
		waitTiDBNotReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 15*time.Minute, interval)

		err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
			err := infraCli.Get(ctx, types.NamespacedName{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())
		waitTiDBReady(ctx, *cluster.Status.SqlEndpoints.PublicEndpoint, 15*time.Minute, interval)

		By("Test active feature, LastActiveTimestamp should be updated")
		duration := time.Second * 10
		err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			// temperately reduce this to test.
			cluster.Annotations[utils.DetectActiveDurationAnnKey] = duration.String()
			err = infraCli.Update(ctx, cluster)
			return err
		})
		Expect(err).Should(Succeed())

		lastActive := *cluster.Status.LastActiveTimestamp
		db, err := sql.Open("mysql", DSN(*cluster.Status.SqlEndpoints.PublicEndpoint))
		Expect(err).Should(Succeed())
		defer db.Close()
		// make a write and check LastActiveTimestamp updated
		err = initRecordTable(ctx, db)
		Expect(err).Should(Succeed())
		err = writeRecord(ctx, db, 1)
		Expect(err).Should(Succeed())
		Eventually(func() error {
			err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
			if err != nil {
				return err
			}
			nowActive := *cluster.Status.LastActiveTimestamp
			updated := nowActive.After(lastActive.Time)
			if !updated {
				return errors.New("LastActiveTimestamp not updated")
			}
			return nil
		}, 2*time.Minute, interval).Should(Succeed())

		By("Test cluster connection with the cloud_admin user")
		dbAdmin, err := sql.Open("mysql", DSNCloudAdmin(*cluster.Status.SqlEndpoints.PublicEndpoint, ""))
		Expect(err).Should(Succeed())
		defer dbAdmin.Close()
		Expect(dbAdmin.PingContext(ctx)).Should(Succeed())

		// By("Test throttler")
		// err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		// 	err := infraCli.Get(ctx, types.NamespacedName{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())
		// lastInfo, err := getTrafficUsageInfo(cluster)
		// Expect(err).Should(Succeed())
		// limitSpeed := 10 * 1024
		// // ref https://github.com/tidbcloud/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())

		// var addedIngress float64
		// var nowInfo *controllercluster.TrafficUsageInfo
		// Eventually(func() error {
		// 	err := infraCli.Get(ctx, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 	if err != nil {
		// 		return err
		// 	}
		// 	nowInfo, err = getTrafficUsageInfo(cluster)
		// 	Expect(err).Should(Succeed())
		// 	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, types.NamespacedName{Namespace: cluster.Namespace, Name: cluster.Name}, cluster)
		// 		if err != nil {
		// 			return err
		// 		}
		// 		cluster.Annotations[providertypes.AnnoKeyIngress] = humanize.IBytes(uint64(limitSpeed)) + "/s"
		// 		err = infraCli.Update(ctx, cluster)
		// 		return err
		// 	})
		// 	Expect(err).Should(Succeed())

		// 	// 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())

		// 	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(infraCli.Delete(ctx, cluster)).Should(Succeed())
		waitClusterDeleted(ctx, cluster, 5*time.Minute, interval)
	})

})

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

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 checkClusterResoucesInEKSDeleted(ctx context.Context, cli client.Client, cluster *cloudv1.Cluster) error {
	tcKey := clusterKeyToTCKey(client.ObjectKeyFromObject(cluster))

	// check tc deleted
	var tcList v1alpha1.TidbClusterList
	err := cli.List(ctx, &tcList, client.InNamespace(tcKey.Namespace))
	if err != nil {
		return err
	}

	for _, item := range tcList.Items {
		if item.DeletionTimestamp != nil {
			continue
		}
		if strings.Contains(item.Name, tcKey.Name) {
			return errors.Errorf("tc %q still exists", item.Name)
		}
	}

	// check tm deleted
	var tmList v1alpha1.TidbMonitorList
	err = cli.List(ctx, &tmList, client.InNamespace(tcKey.Namespace))
	if err != nil {
		return err
	}

	for _, item := range tmList.Items {
		if item.DeletionTimestamp != nil {
			continue
		}
		if strings.Contains(item.Name, tcKey.Name) {
			return errors.Errorf("tm %q still exists", item.Name)
		}
	}

	// check pvc deleted
	var pvcList corev1.PersistentVolumeClaimList
	err = cli.List(ctx, &pvcList, client.InNamespace(tcKey.Namespace))
	if err != nil {
		return err
	}

	for _, item := range pvcList.Items {
		if item.DeletionTimestamp != nil {
			continue
		}
		if strings.Contains(item.Name, tcKey.Name) {
			return errors.Errorf("pvc %q still exists", item.Name)
		}
	}

	// check cert deleted
	var certList cmv1.CertificateList
	err = cli.List(ctx, &certList, client.InNamespace(tcKey.Namespace))
	if err != nil {
		return err
	}

	for _, item := range certList.Items {
		if item.DeletionTimestamp != nil {
			continue
		}
		if strings.Contains(item.Name, tcKey.Name) {
			return errors.Errorf("cert %q still exists", item.Name)
		}
	}

	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
// }