// 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"
	"strings"
	"time"

	"github.com/juju/errors"
	. "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/volumeprovisioner"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	"github.com/pingcap/tidb-operator/test/e2e/util"
	corev1 "k8s.io/api/core/v1"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
)

func testRetention(f *framework.Framework) {
	clusterName := "e2e-retention"
	opts := label.ClusterListOptions(clusterName)
	var pdCount int32 = 5
	var tidbCount int32 = 1
	var tikvCount int32 = 9

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

	pdMap := make(map[string]string)
	tikvSlice := make([]string, 0)
	podList, err := f.KubeCli.CoreV1().Pods(e2eNamespace).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(label.New().Cluster(clusterName).Labels()).String(),
	})
	for _, pod := range podList.Items {
		app := pod.Labels[label.AppLabelKey]
		var pvcName string
		for _, vol := range pod.Spec.Volumes {
			if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
				pvcName = vol.PersistentVolumeClaim.ClaimName
				break
			}
		}
		// tidb
		if len(pvcName) == 0 {
			continue
		}
		pvc, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).Get(pvcName, metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())
		nodeAndHostPath := pvc.Annotations[volumeprovisioner.AnnProvisionerNode] + ":" + pvc.Annotations[volumeprovisioner.AnnProvisionerHostPath]
		if app == label.PDLabelVal {
			pdMap[pod.GetName()] = nodeAndHostPath
		} else if app == label.TiKVLabelVal {
			tikvSlice = append(tikvSlice, nodeAndHostPath)
		}
	}
	util.Logf("pdMap: %v", pdMap)
	util.Logf("tikvSlice: %v", tikvSlice)

	tidbSvcName := fmt.Sprintf("%s-tidb", clusterName)
	svc, err := f.KubeCli.CoreV1().Services(e2eNamespace).Get(tidbSvcName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeLoadBalancer))
	tidbSvcIP := svc.Spec.ClusterIP

	By("When graceful delete TidbCluster")
	tc, err := f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tc.Spec.State = v1.StateGracefulDeleted

	tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
	Expect(err).NotTo(HaveOccurred())

	By("Then wait at most 20 minutes, all resources except PVs should been deleted")
	err = wait.Poll(5*time.Second, 20*time.Minute, func() (bool, error) {
		ts, err := f.Cli.PingcapV1().TidbSets(e2eNamespace).List(opts)
		if err != nil || len(ts.Items) != 0 {
			return false, nil
		}

		cm, err := f.KubeCli.CoreV1().ConfigMaps(e2eNamespace).List(opts)
		if err != nil || len(cm.Items) != 0 {
			return false, nil
		}

		svc, err := f.KubeCli.CoreV1().Services(e2eNamespace).List(opts)
		if err != nil || len(svc.Items) != 1 {
			return false, nil
		}
		if svc.Items[0].Name != tidbSvcName {
			return false, nil
		}

		podList, err := f.KubeCli.CoreV1().Pods(e2eNamespace).List(opts)
		if err != nil || len(podList.Items) != 0 {
			return false, nil
		}

		pvcList, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).List(opts)
		if err != nil || len(pvcList.Items) != 0 {
			return false, nil
		}

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

	By("And wait 1 minutes, TidbCluster and all PVs should not have been deleted")
	err = wait.Poll(5*time.Second, 1*time.Minute, func() (bool, error) {
		pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(opts)
		if err != nil {
			return false, nil
		}

		if len(pvList.Items) != int(pdCount+tikvCount) {
			return false, errors.New("some PVs have been deleted")
		}

		_, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
		if err != nil && apierrs.IsNotFound(err) {
			return false, errors.New("but TidbCluster object had been deleted")
		}

		return false, nil
	})
	if err != nil {
		if !strings.Contains(err.Error(), wait.ErrWaitTimeout.Error()) {
			Expect(err).NotTo(HaveOccurred())
		}
	}

	close(done)
	done = watchRestartCount(f, e2eNamespace, clusterName)
	By("When we restoring the cluster")
	tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tc.Spec.State = v1.StateRestore
	tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
	Expect(err).NotTo(HaveOccurred())

	// set to other state again and again
	for i := 0; i < 3; i++ {
		err := wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			time.Sleep(2 * time.Second)
			tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			tc.Spec.State = v1.StateNormal
			tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
			if err != nil {
				util.Logf(err.Error())
				return false, nil
			}

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

		err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			time.Sleep(3 * time.Second)
			tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			tc.Spec.State = v1.StateGracefulDeleted
			tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
			if err != nil {
				util.Logf(err.Error())
				return false, nil
			}

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

	// ensure finally set to StateRestore
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
		if err != nil {
			return false, nil
		}
		tc.Spec.State = v1.StateRestore
		tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
		if err != nil {
			util.Logf(err.Error())
			return false, nil
		}

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

	By("Then all resources will be ready again")
	allResourcesIsReady(f, e2eNamespace, clusterName, pdCount, 1, tikvCount)

	pdPodList, err := f.KubeCli.CoreV1().Pods(e2eNamespace).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).PD().Labels(),
		).String(),
	})

	podNames := strings.Split(tc.Annotations[v1.AnnPDPodNames], ",")
	Expect(len(podNames)).To(Equal(len(pdPodList.Items)))

	for _, podName := range podNames {
		found := false
		for _, pod := range pdPodList.Items {
			if pod.Name == podName {
				found = true
				break
			}
		}
		Expect(found).To(Equal(true))
	}

	By("And all pods get their PVs")
	podList, err = f.KubeCli.CoreV1().Pods(e2eNamespace).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(label.New().Cluster(clusterName).Labels()).String(),
	})
	for _, pod := range podList.Items {
		app := pod.Labels[label.AppLabelKey]
		var pvcName string
		for _, vol := range pod.Spec.Volumes {
			if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
				pvcName = vol.PersistentVolumeClaim.ClaimName
				break
			}
		}
		// tidb
		if len(pvcName) == 0 {
			continue
		}
		pvc, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).Get(pvcName, metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())
		pv, err := f.KubeCli.CoreV1().PersistentVolumes().Get(pvc.Spec.VolumeName, metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())
		pvcNodeAndHostPath := pvc.Annotations[volumeprovisioner.AnnProvisionerNode] + ":" + pvc.Annotations[volumeprovisioner.AnnProvisionerHostPath]
		pvNodeAndHostPath := pv.Annotations[volumeprovisioner.AnnProvisionerNode] + ":" + pv.Annotations[volumeprovisioner.AnnProvisionerHostPath]
		util.Logf(pod.GetName())
		if app == label.PDLabelVal {
			Expect(pdMap[pod.GetName()]).To(Equal(pvcNodeAndHostPath))
			Expect(pdMap[pod.GetName()]).To(Equal(pvNodeAndHostPath))
		} else if app == label.TiKVLabelVal {
			found := false
			for idx, nap := range tikvSlice {
				if nap == pvcNodeAndHostPath && nap == pvNodeAndHostPath {
					found = true
					tikvSlice = append(tikvSlice[:idx], tikvSlice[idx+1:]...)
					break
				}
			}
			Expect(found).To(Equal(true))
		}
	}

	By("And TidbCluster.Spec.State will be set to StateNormal")
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		tc, err := f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
		if err != nil {
			return false, nil
		}

		if tc.Spec.State != v1.StateNormal {
			return false, nil
		}

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

	By("And Ensure LoadBalancer service is the same as before")
	svc, err = f.KubeCli.CoreV1().Services(e2eNamespace).Get(tidbSvcName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())
	Expect(svc.Spec.ClusterIP).To(Equal(tidbSvcIP))

	svcName := fmt.Sprintf("%s-tidb", clusterName)
	tableName := fixTableName(clusterName)
	ensureDataIsRight(svcName, e2eNamespace, "test", tableName, clusterName)

	By("When graceful delete TidbCluster again")
	tc, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tc.Spec.State = v1.StateGracefulDeleted
	_, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Update(tc)
	Expect(err).NotTo(HaveOccurred())

	By("Then wait 1 minutes, TidbCluster and its PVs should not be deleted")
	err = wait.Poll(5*time.Second, 1*time.Minute, func() (bool, error) {
		pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(opts)
		if err != nil {
			return false, nil
		}

		if len(pvList.Items) != int(pdCount+tikvCount) {
			return true, errors.Errorf("PV count not equal: %d", pdCount+tikvCount)
		}

		_, err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
		if err != nil && apierrs.IsNotFound(err) {
			return true, errors.Errorf("TidbCluster [%s/%s] had been deleted", e2eNamespace, clusterName)
		}

		return false, nil
	})
	if err != nil {
		if !strings.Contains(err.Error(), wait.ErrWaitTimeout.Error()) {
			Expect(err).NotTo(HaveOccurred())
		}
	}

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