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

package e2e

import (
	"fmt"
	"os/exec"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	tsctl "github.com/pingcap/tidb-operator/pkg/tidbset/controller"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	"github.com/pingcap/tidb-operator/test/e2e/util"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
)

const (
	zkPodName    = "zk-0"
	kafkaPodName = "kafka-0"
)

func testBinlog(f *framework.Framework) {
	var tidbCount int32 = 2
	var pdCount int32 = 1
	var tikvCount int32 = 3
	clusterName := "e2e-binlog"
	binlogImage := "localhost:5000/pingcap/tidb-binlog:latest"
	upgradeBinlogImage := "uhub.ucloud.cn/pingcap/tidb-binlog:latest"

	By("Ensure kafka is ready")
	time.Sleep(10 * time.Second)
	ensureKafkaIsReady(f, e2eNamespace)

	// TODO(tennix): this only make sure TiDB with binlog is up and running
	// we should write some data to TiDB and ensure binlog is written into kafka
	done := deployCluster(f, e2eNamespace, clusterName, pdCount, tidbCount, tikvCount)

	By("Ensure TiDB has not enabled binlog")
	tidbPL, err := f.GetAllTiDBPods(e2eNamespace, clusterName)
	Expect(err).NotTo(HaveOccurred())
	Expect(hasAllBinlogEnabled(tidbPL)).NotTo(Equal(true))

	By("When upgrade TiDB to support binlog")
	tc, err := f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tc.Spec.TiDB.Binlog = &v1.ContainerSpec{
		Image: binlogImage,
	}
	tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
	Expect(err).NotTo(HaveOccurred())

	By("Then all TiDB pods should upgrade to contain binlog container")
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		tidbPL, err = f.GetAllTiDBPods(e2eNamespace, clusterName)
		if err != nil || len(tidbPL.Items) != int(tidbCount) {
			return false, nil
		}

		if !hasAllBinlogEnabled(tidbPL) {
			return false, nil
		}

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

			blImage := tsctl.GetImageByPodSpec(v1.BinlogContainerType, &pod.Spec)
			if binlogImage != blImage {
				return false, nil
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("When upgrade Binlog image to %s", upgradeBinlogImage))
	tc.Spec.TiDB.Binlog.Image = upgradeBinlogImage
	_, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then all Binlog containers along with TiDB pods are upgraded to %s", upgradeBinlogImage))
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		tidbPL, err = f.GetAllTiDBPods(e2eNamespace, clusterName)
		if err != nil || len(tidbPL.Items) != int(tidbCount) {
			return false, nil
		}

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

			blImage := tsctl.GetImageByPodSpec(v1.BinlogContainerType, &pod.Spec)
			if upgradeBinlogImage != blImage {
				return false, nil
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By("When create a standalone tidb")
	cmd := exec.Command("/bin/sh", "-c",
		fmt.Sprintf("/usr/local/bin/kubectl apply -n %s -f /tmp/data/drainer/tidb.yaml", e2eNamespace))
	data, err := cmd.CombinedOutput()
	if err != nil {
		util.Logf(string(data))
	}
	Expect(err).NotTo(HaveOccurred())

	By("Then temporary tidb should be created successfully")
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		pod, err := f.KubeCli.CoreV1().Pods(e2eNamespace).Get("standalone-tidb", metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		if pod.Status.Phase != corev1.PodRunning {
			return false, nil
		}
		_, err = f.KubeCli.CoreV1().Services(e2eNamespace).Get("standalone-tidb", metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	ensureDrainerIsReady(f, e2eNamespace)

	svcName := "standalone-tidb"
	tableName := fixTableName(svcName)
	val := "abc"
	addDataToTiDB(svcName, e2eNamespace, tableName, val)
	ensureDataIsRight(svcName, e2eNamespace, "test", tableName, val)

	deleteCluster(f, e2eNamespace, clusterName, done, pdCount+tikvCount)
}

// deploy one zk + one kafka for test
func ensureKafkaIsReady(f *framework.Framework, ns string) {
	cmd := exec.Command("/bin/sh", "-c", "/usr/local/bin/kubectl apply -f /tmp/data/kafka/zookeeper.yaml")
	data, err := cmd.CombinedOutput()
	if err != nil {
		util.Logf("kubectl apply zookeeper.yaml failed,error: %s", string(data))
	}
	Expect(err).NotTo(HaveOccurred())
	Expect(waitPodReady(f, ns, zkPodName)).NotTo(HaveOccurred())

	cmd = exec.Command("/bin/sh", "-c", "/usr/local/bin/kubectl apply -f /tmp/data/kafka/kafka.yaml")
	data, err = cmd.CombinedOutput()
	if err != nil {
		util.Logf("kubectl apply kafka.yaml failed,error: %s", string(data))
	}
	Expect(err).NotTo(HaveOccurred())
	Expect(waitPodReady(f, ns, kafkaPodName)).NotTo(HaveOccurred())
}

func ensureDrainerIsReady(f *framework.Framework, ns string) {
	cmd := fmt.Sprintf("helm install /tmp/drainer --namespace=%s --name=tidb-drainer --values=/tmp/data/drainer/values.yaml", ns)
	data, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
	}
	Expect(err).NotTo(HaveOccurred())

	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		podList, err := f.KubeCli.CoreV1().Pods(ns).List(metav1.ListOptions{
			LabelSelector: "app=tidb-drainer",
		})
		if err != nil {
			return false, nil
		}
		if len(podList.Items) != 1 {
			return false, nil
		}
		pod := podList.Items[0]
		if pod.Status.Phase != corev1.PodRunning {
			return false, nil
		}
		for _, cond := range pod.Status.Conditions {
			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
				return true, nil
			}
		}

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

func waitPodReady(f *framework.Framework, ns, podName string) error {
	return wait.Poll(10*time.Second, 10*time.Minute, func() (bool, error) {
		pod, err := f.KubeCli.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
		if err != nil || pod.Status.Phase != corev1.PodRunning {
			return false, nil
		}
		for _, cond := range pod.Status.Conditions {
			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
				return true, nil
			}
		}
		return false, nil
	})
}

func hasAllBinlogEnabled(tidbPL *corev1.PodList) bool {
	binlogEnabled := true
	for _, tidbPod := range tidbPL.Items {
		if len(tidbPod.Spec.Containers) == 1 {
			binlogEnabled = false
			break
		}
	}
	return binlogEnabled
}