// Copyright 2017 PingCAP, Inc.
//
// Licensed 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,
// See the License for the specific language governing permissions and
// limitations under the License.package spec

package e2e

import (
	"bufio"
	"database/sql"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"

	_ "github.com/go-sql-driver/mysql" // its just a test
	"github.com/juju/errors"
	"github.com/ngaut/log"
	. "github.com/onsi/ginkgo"
	"github.com/onsi/ginkgo/config"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/test/e2e/bootstrap"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	"github.com/pingcap/tidb-operator/test/e2e/util"

	"testing"

	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
)

const (
	e2eNamespace       = "e2e-ns"
	e2eNamespaceDelete = "e2e-ns-delete"
)

var bs *bootstrap.Bootstrap
var writer *bufio.Writer

func init() {
	image := os.Getenv("OPERATOR_IMAGE")
	if image == "" {
		log.Fatal("OPERATOR_IMAGE environment variable not set")
	}
	bootstrap.OperatorImage = image
}

func TestE2E(t *testing.T) {
	RegisterFailHandler(failWrapper)
	logPath := fmt.Sprintf("/logdir/%d", config.GinkgoConfig.RandomSeed)
	if _, err := os.Stat(logPath); os.IsNotExist(err) {
		err = os.MkdirAll(logPath, os.ModePerm)
		if err != nil {
			panic(err)
		}
	}
	logFile, err := os.Create(filepath.Join(logPath, fmt.Sprintf("%d", config.GinkgoConfig.ParallelNode)))
	if err != nil {
		panic(err)
	}
	writer = bufio.NewWriter(logFile)

	RunSpecs(t, "TiDB Operator Smoke tests")
}

var _ = Describe("Smoke", func() {
	f := framework.New("tidbcluster")

	It("should increase tidb cluster [Increase]", func() {
		testIncrease(f)
	})

	It("should decrease tidb cluster [Decrease]", func() {
		testDecrease(f)
	})

	It("should pause tidb cluster [Paused]", func() {
		testPause(f)
	})

	It("should upgrade tidb cluster [Upgrade]", func() {
		testUpgrade(f)
	})

	It("should works as normal if we delete a PV used by a tikv instance [Volume]", func() {
		testVolume(f)
	})

	It("should delete PVs when delete namespace [DeleteNamespace]", func() {
		testDeleteNamespace(f)
	})

	It("should retention PVs [Retention]", func() {
		testRetention(f)
	})

	It("should Sync all required resources [Sync]", func() {
		testSync(f)
	})

	It("should deploy monitor and privileged-tidb as Deployment [Monitor]", func() {
		testMonitor(f)
	})

	It("should deploy tidb with binlog [Binlog]", func() {
		testBinlog(f)
	})

	It("should import tidb data with lightning [Lightning]", func() {
		testLightning(f)
	})

	It("should backup tidb with [Backup]", func() {
		testBackup(f)
	})

	It("should be allocated on demand when there are many tidb volumes fittable [Serial,BestFit]", func() {
		testVolumeBestFit(f)
	})

	It("should work as expected [Serial,HA]", func() {
		testHA(f)
	})
})

var _ = SynchronizedBeforeSuite(func() []byte {
	By("Bootstrapping TiDB Operator")
	var err error

	bs, err = bootstrap.New()
	Expect(err).NotTo(HaveOccurred())
	approveAllChan := approveReviews(bs.Cli, "")

	Expect(bs.ResetE2E([]string{e2eNamespace, e2eNamespaceDelete})).To(Succeed())

	err = bs.DeleteOperator()
	if err != nil {
		util.Logf(err.Error())
	}
	Expect(bs.WaitOperatorDeleted()).To(Succeed())
	approveReviews(bs.Cli, "e2e-sync")
	Expect(bs.ApplyOperator()).To(Succeed())
	Expect(bs.WaitOperatorReady()).To(Succeed())

	Expect(bs.SetupTidbVolumeConfigMap()).To(Succeed())
	Expect(bs.ClearPVs()).To(Succeed())

	close(approveAllChan)
	return []byte(nil)
}, func([]byte) {})

var _ = SynchronizedAfterSuite(func() {
	By("Tearing down TiDB Operator")
	err := bs.Teardown()
	Expect(err).NotTo(HaveOccurred())
}, func() {})

func fixTableName(name string) string {
	return strings.Replace(name, "-", "_", -1)
}

func deployCluster(f *framework.Framework, ns, clusterName string, pdCount, tidbCount, tikvCount int32) chan struct{} {
	By(fmt.Sprintf("When tidb cluster: %s/%s was created", ns, clusterName))
	cmd := fmt.Sprintf("helm install /tmp/tidb-cluster --namespace=%s --name=%s --values=/tmp/data/%s-values.yaml", ns, clusterName, clusterName)
	data, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
	}
	Expect(err).NotTo(HaveOccurred())

	done := watchRestartCount(f, ns, clusterName)

	allResourcesIsReady(f, ns, clusterName, pdCount, tidbCount, tikvCount)

	By(fmt.Sprintf("When add data to tidb"))
	svcName := fmt.Sprintf("%s-tidb", clusterName)
	tableName := fixTableName(clusterName)
	addDataToTiDB(svcName, ns, tableName, clusterName)
	ensureDataIsRight(svcName, ns, "test", tableName, clusterName)

	return done
}

func addDataToTiDB(svcName, ns, tableName, val string) {
	err := wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
		db, err := sql.Open("mysql", fmt.Sprintf("root:@(%s.%s:4000)/test?charset=utf8", svcName, ns))
		if err != nil {
			util.Logf("can't open connection to mysql: %v", err)
			return false, nil
		}
		defer db.Close()

		_, err = db.Exec(fmt.Sprintf("CREATE TABLE %s (name VARCHAR(64))", tableName))
		if err != nil {
			util.Logf("can't create table to mysql: %v", err)
			return false, nil
		}

		_, err = db.Exec(fmt.Sprintf("INSERT INTO %s VALUES (?)", tableName), val)
		if err != nil {
			util.Logf("can't insert data to mysql: %v", err)
			return false, nil
		}

		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())
}

func ensureDataIsRight(svcName, ns, dbName, tableName, val string) {
	By("Ensure saved data is right")
	err := wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
		db, err := sql.Open("mysql", fmt.Sprintf("root:@(%s.%s:4000)/%s?charset=utf8", svcName, ns, dbName))
		if err != nil {
			return false, nil
		}

		rows, err := db.Query(fmt.Sprintf("SELECT * FROM %s", tableName))
		if err != nil {
			util.Logf(err.Error())
			return false, nil
		}

		for rows.Next() {
			var v string
			err := rows.Scan(&v)
			if err != nil {
				util.Logf(err.Error())
			}

			if v == val {
				return true, nil
			}

			return true, errors.Errorf("val should equal: %s", val)
		}

		return false, nil
	})
	Expect(err).NotTo(HaveOccurred())
}

func allResourcesIsReady(f *framework.Framework, ns, clusterName string, pdCount, tidbCount, tikvCount int32) {
	By("Then all its members are running")
	err := wait.Poll(5*time.Second, 30*time.Minute, func() (bool, error) {
		podList, err := f.GetAllClusterPods(ns, clusterName)
		if err != nil {
			return false, nil
		}

		if len(podList.Items) < int(pdCount+tidbCount+tikvCount) {
			return false, nil
		}

		for _, pod := range podList.Items {
			running, err := f.IsPodRunning(&pod)
			if err != nil || !running {
				return false, nil
			}

		}

		return true, err
	})
	Expect(err).NotTo(HaveOccurred())

	tc, err := f.Cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	if tc.Spec.Config != nil {
		By("And its cluster configmap is created")
		_, err = f.KubeCli.CoreV1().ConfigMaps(ns).Get(fmt.Sprintf("%s-config", clusterName), metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())
	}

	By("And its services are created")
	_, err = f.KubeCli.CoreV1().Services(ns).Get(fmt.Sprintf("%s-tidb", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	_, err = f.KubeCli.CoreV1().Services(ns).Get(fmt.Sprintf("%s-pd", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	_, err = f.KubeCli.CoreV1().Services(ns).Get(fmt.Sprintf("%s-pd-peer", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	_, err = f.KubeCli.CoreV1().Services(ns).Get(fmt.Sprintf("%s-tikv-peer", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	By("And its TidbSets are created")
	_, err = f.Cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-pdset", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	_, err = f.Cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-tidbset", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	_, err = f.Cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-tikvset", clusterName), metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	By("And all tikv instances's PVCs are created")
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		pvcList, err := f.KubeCli.CoreV1().PersistentVolumeClaims(ns).List(label.ClusterListOptions(clusterName))
		if err != nil {
			return false, nil
		}

		if len(pvcList.Items) != int(pdCount+tikvCount) {
			return false, nil
		}

		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By("And cluster's PVs are created")
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(label.ClusterListOptions(clusterName))
		if err != nil {
			return false, nil
		}

		if len(pvList.Items) != int(pdCount+tikvCount) {
			return false, nil
		}

		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())
}

func deleteCluster(f *framework.Framework, ns, clusterName string, done chan struct{}, pvCount int32) {
	By(fmt.Sprintf("When deleting tidb cluster: %s/%s", ns, clusterName))
	data, err := exec.Command("/bin/sh", "-c", fmt.Sprintf("helm delete --purge %s", clusterName)).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
	}

	By("Then all its members are deleted")
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		podList, err := f.GetAllClusterPods(ns, clusterName)
		if err != nil {
			return false, nil
		}
		if len(podList.Items) != 0 {
			return false, nil
		}

		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	pvShouldNotBeDeleted(f, clusterName, pvCount)
	close(done)
}

func pvShouldNotBeDeleted(f *framework.Framework, clusterName string, pvCount int32) {
	By("Then wait 1 minutes, pvs should not been deleted")
	time.Sleep(1 * time.Minute)
	opts := label.ClusterListOptions(clusterName)
	pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(opts)
	Expect(err).NotTo(HaveOccurred())

	nowPVCount := len(pvList.Items)
	if nowPVCount != int(pvCount) {
		Expect(errors.Errorf("pv count is not: %d, now: %d", pvCount, nowPVCount)).NotTo(HaveOccurred())
	}
}

func approveReviews(cli versioned.Interface, except string) chan struct{} {
	done := make(chan struct{})

	go func() {
		defer GinkgoRecover()
	Loop:
		for {
			select {
			case <-done:
				break Loop
			case <-time.After(10 * time.Second):
				opts := metav1.ListOptions{
					LabelSelector: labels.SelectorFromSet(
						label.New().Labels(),
					).String(),
				}
				reviewList, err := cli.PingcapV1().TidbReviews().List(opts)
				if err != nil {
					util.Logf(err.Error())
					continue
				}

				for _, r := range reviewList.Items {
					clusterName := r.Labels[label.ClusterLabelKey]
					shouldApprove := true
					if except != "" && clusterName == except {
						shouldApprove = false
					}
					if shouldApprove && r.Status.Approved != true {
						util.Logf("setting tidbReview [%s]: clusterName: [%s], reason: [%s], message: [%s] to approved",
							r.GetName(), clusterName, r.Spec.Reason, r.Spec.Message)
						r.Status.Approved = true
						_, err := cli.PingcapV1().TidbReviews().Update(&r)
						if err != nil {
							util.Logf(err.Error())
							continue
						}
					}
				}
			}
		}

		util.Logf("approveReview goroutine exit")
	}()

	return done
}

func watchRestartCount(f *framework.Framework, ns, clusterName string) chan struct{} {
	done := make(chan struct{})
	maxRestartCount := int32(3)

	go func() {
		defer GinkgoRecover()
	Loop:
		for {
			select {
			case <-done:
				break Loop
			case <-time.After(10 * time.Millisecond):
				podList, err := f.GetAllClusterPods(ns, clusterName)
				if err != nil {
					continue
				}

				for _, pod := range podList.Items {
					for _, cs := range pod.Status.ContainerStatuses {
						if cs.RestartCount > maxRestartCount {
							failWrapper(fmt.Sprintf("POD: %s/%s's container: %s's restartCount is greater than: %d", pod.GetNamespace(), pod.GetName(), cs.Name, maxRestartCount))
						}
					}
				}
			}
		}
	}()

	return done
}

func failWrapper(message string, callerSkip ...int) {
	defer Fail(message, callerSkip...)

	dumpLog("kubectl get po -owide -n kube-system")
	dumpLog("kubectl get po -owide -n pingcap")
	dumpLog(fmt.Sprintf("kubectl get po,pvc,svc,cm,tidbclusters,tidbsets -owide -n %s", e2eNamespace))
	dumpLog(fmt.Sprintf("kubectl get po,pvc,svc,cm,tidbclusters,tidbsets -owide -n %s", e2eNamespaceDelete))
	dumpLog(fmt.Sprintf("kubectl get po,pvc,svc,cm,tidbclusters,tidbsets -n %s -oyaml", e2eNamespace))
	dumpLog(fmt.Sprintf("kubectl get po,pvc,svc,cm,tidbclusters,tidbsets -n %s -oyaml", e2eNamespaceDelete))
	dumpLog("kubectl get pv")
	dumpLog("kubectl get pv -oyaml")

	cfg, err := rest.InClusterConfig()
	if err != nil {
		util.Logf(err.Error())
		return
	}
	kubeCli, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		util.Logf(err.Error())
		return
	}

	pingcapPodList, err := kubeCli.CoreV1().Pods("pingcap").List(metav1.ListOptions{})
	if err != nil {
		util.Logf(err.Error())
		return
	}
	for _, pod := range pingcapPodList.Items {
		dumpLog(fmt.Sprintf("kubectl -n pingcap get po  %s -o yaml", pod.GetName()))
		for _, c := range pod.Spec.Containers {
			dumpLog(fmt.Sprintf("kubectl logs -n pingcap %s -c %s", pod.GetName(), c.Name))
			dumpLog(fmt.Sprintf("kubectl logs -n pingcap %s -c %s -p", pod.GetName(), c.Name))
		}
	}

	e2ePodList, err := kubeCli.CoreV1().Pods(e2eNamespace).List(metav1.ListOptions{})
	if err != nil {
		util.Logf(err.Error())
		return
	}
	for _, pod := range e2ePodList.Items {
		dumpLog(fmt.Sprintf("kubectl -n %s get po  %s -o yaml", e2eNamespace, pod.GetName()))
		for _, c := range pod.Spec.Containers {
			dumpLog(fmt.Sprintf("kubectl logs -n %s %s -c %s", e2eNamespace, pod.GetName(), c.Name))
			dumpLog(fmt.Sprintf("kubectl logs -n %s %s -c %s -p", e2eNamespace, pod.GetName(), c.Name))
		}
	}

	e2eDeletePodList, err := kubeCli.CoreV1().Pods(e2eNamespaceDelete).List(metav1.ListOptions{})
	if err != nil {
		util.Logf(err.Error())
		return
	}
	for _, pod := range e2eDeletePodList.Items {
		dumpLog(fmt.Sprintf("kubectl -n %s get po  %s -o yaml", e2eNamespaceDelete, pod.GetName()))
		for _, c := range pod.Spec.Containers {
			dumpLog(fmt.Sprintf("kubectl logs -n %s %s -c %s", e2eNamespaceDelete, pod.GetName(), c.Name))
			dumpLog(fmt.Sprintf("kubectl logs -n %s %s -c %s -p", e2eNamespaceDelete, pod.GetName(), c.Name))
		}
	}

}

func dumpLog(cmdStr string) {
	writer.WriteString(fmt.Sprintf("$ %s\n", cmdStr))
	data, err := exec.Command("/bin/sh", "-c", "/usr/local/bin/"+cmdStr).CombinedOutput()
	if err != nil {
		writer.WriteString(err.Error())
	}
	writer.WriteString(string(data))
}