// 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 (
	"fmt"
	"os/exec"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/pkg/util/marshal"
	"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/labels"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/apimachinery/pkg/util/wait"
	clientsetscheme "k8s.io/client-go/kubernetes/scheme"
)

var codecs serializer.CodecFactory

func init() {
	v1.AddToScheme(clientsetscheme.Scheme)
	codecs = clientsetscheme.Codecs
}

func testSync(f *framework.Framework) {
	clusterName := "e2e-sync"
	var pdCount int32 = 3
	var tikvCount int32 = 3
	var tidbCount int32 = 1

	done := deployCluster(f, e2eNamespace, clusterName, pdCount, tidbCount, tikvCount)

	podListOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).Labels(),
		).String(),
	}

	By("Then pod cluster-id, member-id, store-id labels are synced.")
	err := wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		podList, err := f.KubeCli.CoreV1().Pods(e2eNamespace).List(podListOptions)
		if err != nil {
			return false, nil
		}
		for _, pod := range podList.Items {
			ls := pod.GetLabels()
			util.Logf("pod[%s] clusterID=%s", pod.GetName(), ls[label.ClusterIDLabelKey])
			if ls[label.ClusterIDLabelKey] == "" {
				return false, nil
			}
			switch ls[label.AppLabelKey] {
			case label.PDLabelVal:
				util.Logf("pod[%s] memberID=%s", pod.GetName(), ls[label.MemberIDLabelKey])
				if ls[label.MemberIDLabelKey] == "" {
					return false, nil
				}
			case label.TiKVLabelVal:
				util.Logf("pod[%s] storeID=%s", pod.GetName(), ls[label.StoreIDLabelKey])
				if ls[label.StoreIDLabelKey] == "" {
					return false, nil
				}
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By("And PVC cluster-id, member-id, store-id labels are synced.")
	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
		}
		for _, pv := range pvList.Items {
			ls := pv.GetLabels()
			util.Logf("pv[%s] clusterID=%s", pv.GetName(), ls[label.ClusterIDLabelKey])
			if ls[label.ClusterIDLabelKey] == "" {
				return false, nil
			}
			if ls[label.AppLabelKey] == label.PDLabelVal {
				util.Logf("pv[%s] memberID=%s", pv.GetName(), ls[label.MemberIDLabelKey])
				if ls[label.MemberIDLabelKey] == "" {
					return false, nil
				}
			} else if ls[label.AppLabelKey] == label.TiKVLabelVal {
				util.Logf("pv[%s] storeID=%s", pv.GetName(), ls[label.StoreIDLabelKey])
				if ls[label.StoreIDLabelKey] == "" {
					return false, nil
				}
			}
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("When remove PVC's labels: %s and annotations: %s, remove PV's labels: %s, %s, and annotations: %s", label.AppLabelKey, label.AnnPodNameKey, label.AppLabelKey, label.NamespaceLabelKey, label.AnnPodNameKey))
	pvcList, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).List(label.ClusterListOptions(clusterName))
	Expect(err).NotTo(HaveOccurred())
	pvc := &(pvcList.Items[0])

	podList, err := f.KubeCli.CoreV1().Pods(e2eNamespace).List(podListOptions)
	Expect(err).NotTo(HaveOccurred())
	var pod *corev1.Pod
outerLabel:
	for _, p := range podList.Items {
		for _, vol := range p.Spec.Volumes {
			if vol.Name == v1.TiDBVolumeName && vol.VolumeSource.PersistentVolumeClaim != nil && pvc.GetName() == vol.VolumeSource.PersistentVolumeClaim.ClaimName {
				pod = &p
				break outerLabel
			}
		}
	}
	Expect(pod).NotTo(Equal(nil))

	pvcApp := pvc.Labels[label.AppLabelKey]
	util.Logf("PVC: [%s/%s]'s %s: %s", e2eNamespace, pvc.GetName(), label.AppLabelKey, pvcApp)

	pvc.Labels[label.AppLabelKey] = ""
	delete(pvc.Annotations, label.AnnPodNameKey)

	Expect(pvc.Labels[label.AppLabelKey]).To(Equal(""))
	Expect(pvc.Annotations[label.AnnPodNameKey]).To(Equal(""))
	pvc, err = f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).Update(pvc)
	Expect(err).NotTo(HaveOccurred())

	pv, err := f.KubeCli.CoreV1().PersistentVolumes().Get(pvc.Spec.VolumeName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	pvNs := pv.Labels[label.NamespaceLabelKey]
	pvApp := pv.Labels[label.AppLabelKey]

	delete(pv.Labels, label.AppLabelKey)
	pv.Annotations[label.AnnPodNameKey] = ""
	pv.Labels[label.NamespaceLabelKey] = ""

	Expect(pv.Labels[label.NamespaceLabelKey]).To(Equal(""))
	Expect(pv.Annotations[label.AnnPodNameKey]).To(Equal(""))
	Expect(pv.Labels[label.AppLabelKey]).To(Equal(""))
	pv, err = f.KubeCli.CoreV1().PersistentVolumes().Update(pv)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then PVC's labels: %s and annotations: %s must be set again", label.AppLabelKey, label.AnnPodNameKey))
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		pvc, err = f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).Get(pvc.GetName(), metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		if pvc.Labels[label.AppLabelKey] != pvcApp {
			return false, nil
		}
		if pvc.Annotations[label.AnnPodNameKey] != pod.GetName() {
			return false, nil
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("Then PV's labels: %s, %s and annotations: %s must be set again", label.AppLabelKey, label.NamespaceLabelKey, label.AnnPodNameKey))
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		pv, err = f.KubeCli.CoreV1().PersistentVolumes().Get(pv.GetName(), metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		if pv.Labels[label.AppLabelKey] != pvApp {
			return false, nil
		}
		if pv.Labels[label.NamespaceLabelKey] != pvNs {
			return false, nil
		}
		if pv.Annotations[label.AnnPodNameKey] != pod.GetName() {
			return false, nil
		}
		return true, nil
	})
	Expect(err).NotTo(HaveOccurred())

	By("And all tidbclusters, pvs and configmap are already backed up")
	for _, pod := range podList.Items {
		k := pod.Labels[label.AppLabelKey]
		if k != label.PDLabelVal && k != label.TiKVLabelVal {
			continue
		}

		ensureTidbClusterAreBackupUp(f, pod)
		ensurePVsAreBackupUp(f, pod)
		ensureConfigMapAreBackupUp(f, pod)
	}

	By(fmt.Sprintf("And deleting tidb cluster: %s/%s", e2eNamespace, clusterName))
	_, err = exec.Command("/bin/sh", "-c", fmt.Sprintf("helm delete --purge %s", clusterName)).CombinedOutput()
	Expect(err).NotTo(HaveOccurred())

	time.Sleep(2 * time.Minute)
	podList, err = f.KubeCli.CoreV1().Pods(e2eNamespace).List(podListOptions)
	Expect(err).NotTo(HaveOccurred())
	Expect(len(podList.Items)).To(Equal(int(pdCount + tikvCount + tidbCount)))

	// approved all e2e-sync reviews from now
	go func() {
		for {
			opts := label.ClusterListOptions(clusterName)
			reviewList, err := f.Cli.PingcapV1().TidbReviews().List(opts)
			if err != nil {
				util.Logf(err.Error())
				continue
			}

			for _, r := range reviewList.Items {
				if r.Labels[label.ClusterLabelKey] == "e2e-sync" {
					if r.Status.Approved != true {
						util.Logf("[sync] 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 := f.Cli.PingcapV1().TidbReviews().Update(&r)
						if err != nil {
							util.Logf(err.Error())
							continue
						}
					}
				}
			}
		}
	}()

	time.Sleep(5 * time.Minute)
	podList, err = f.KubeCli.CoreV1().Pods(e2eNamespace).List(podListOptions)
	Expect(err).NotTo(HaveOccurred())
	Expect(len(podList.Items)).To(Equal(0))

	pvShouldNotBeDeleted(f, clusterName, pdCount+tikvCount)
	close(done)
}

func ensureTidbClusterAreBackupUp(f *framework.Framework, pod corev1.Pod) {
	k := pod.Labels[label.AppLabelKey]
	cmd := fmt.Sprintf("/usr/local/bin/kubectl exec -n %s %s -c %s cat /var/lib/tidb/%s/k8s-metadata/tidbcluster.yaml",
		e2eNamespace, pod.Name, k, k)
	data, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
		Expect(err).NotTo(HaveOccurred())
	}

	tcObj, err := marshal.UnmarshalFromYamlForCodecs(data, v1.SchemeGroupVersion, codecs)
	if err != nil {
		util.Logf("%s-tc", err.Error())
		Expect(err).NotTo(HaveOccurred())
	}
	Expect(err).NotTo(HaveOccurred())

	tc, ok := tcObj.(*v1.TidbCluster)
	Expect(ok).To(Equal(true))

	Expect(tc.Name).To(Equal(pod.Labels[label.ClusterLabelKey]))
}

func ensurePVsAreBackupUp(f *framework.Framework, pod corev1.Pod) {
	k := pod.Labels[label.AppLabelKey]
	data, err := exec.Command("/bin/sh", "-c",
		fmt.Sprintf("/usr/local/bin/kubectl exec -n %s %s -c %s cat /var/lib/tidb/%s/k8s-metadata/pv.yaml",
			e2eNamespace, pod.Name, k, k)).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
		Expect(err).NotTo(HaveOccurred())
	}

	pvListObj, err := marshal.UnmarshalFromYamlForCodecs(data, corev1.SchemeGroupVersion, codecs)
	if err != nil {
		util.Logf("%s-pv", err.Error())
		Expect(err).NotTo(HaveOccurred())
	}

	pvList, ok := pvListObj.(*corev1.PersistentVolumeList)
	Expect(ok).To(Equal(true))

	for _, pv := range pvList.Items {
		Expect(pv.Labels[label.ClusterLabelKey]).To(Equal(pod.Labels[label.ClusterLabelKey]))
		Expect(pv.Labels[label.NamespaceLabelKey]).To(Equal(pod.Namespace))
	}
}

func ensureConfigMapAreBackupUp(f *framework.Framework, pod corev1.Pod) {
	k := pod.Labels[label.AppLabelKey]
	data, err := exec.Command("/bin/sh", "-c",
		fmt.Sprintf("/usr/local/bin/kubectl exec -n %s %s -c %s cat /var/lib/tidb/%s/k8s-metadata/configmap.yaml",
			e2eNamespace, pod.Name, k, k)).CombinedOutput()
	if err != nil {
		util.Logf(string(data))
		Expect(err).NotTo(HaveOccurred())
	}
}