// 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"
	"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/test/e2e/framework"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
)

func testPause(f *framework.Framework) {
	clusterName := "e2e-pause"
	ns := e2eNamespace
	var pdCount int32 = 1
	var tidbCount int32 = 2
	var tikvCount int32 = 3

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

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

	tc.Spec.Paused = true

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

	By(fmt.Sprintf("Then Annotations[\"%s\"] is true", label.AnnPaused))
	err = ensureAnnotationEqual("true", ns, clusterName, f)
	Expect(err).NotTo(HaveOccurred())

	By("When wait for 2 minutes and delete a TiKV and a TiDB pod")
	time.Sleep(2 * time.Minute)

	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		tikvList, err := f.GetAllTiKVPods(ns, clusterName)
		if err != nil {
			return false, nil
		}

		tikvPod := tikvList.Items[0]
		// pvc should be deleted before tikv pod
		var pvcName string
		for _, vol := range tikvPod.Spec.Volumes {
			if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
				pvcName = vol.PersistentVolumeClaim.ClaimName
				break
			}
		}
		if pvcName == "" {
			return false, nil
		}
		err = f.KubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(pvcName, nil)
		if err != nil {
			return false, nil
		}

		err = f.KubeCli.CoreV1().Pods(ns).Delete(tikvPod.GetName(), nil)
		if err != nil {
			return false, nil
		}

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

	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		tidbList, err := f.GetAllTiDBPods(ns, clusterName)
		if err != nil {
			return false, nil
		}

		tidbPod := tidbList.Items[0]
		err = f.KubeCli.CoreV1().Pods(ns).Delete(tidbPod.GetName(), nil)
		if err != nil {
			return false, nil
		}

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

	By("Then waiting for 1 minutes to make sure pods are not recreated and PV/PVC still exist")
	time.Sleep(1 * time.Minute)
	err = wait.Poll(5*time.Second, 1*time.Minute, func() (bool, error) {
		podList, err := f.GetAllTiKVPods(ns, clusterName)
		if err != nil {
			return false, err
		}

		if len(podList.Items) != int(tikvCount)-1 {
			return false, errors.Errorf("tikv instances is not equal %d, now: %d", tikvCount-1, len(podList.Items))
		}

		podList, err = f.GetAllTiDBPods(ns, clusterName)
		if err != nil {
			return false, err
		}
		if len(podList.Items) != int(tidbCount)-1 {
			return false, errors.Errorf("tidb instances is not equal %d, now: %d", tidbCount-1, len(podList.Items))
		}

		pvcList, err := f.KubeCli.CoreV1().PersistentVolumeClaims(ns).List(label.ClusterListOptions(clusterName))
		if err != nil {
			return false, err
		}
		if len(pvcList.Items) != int(pdCount+tikvCount)-1 {
			return false, errors.Errorf("PVC not equal to %d, now: %d", pdCount+tikvCount-1, len(pvcList.Items))
		}

		pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(label.ClusterListOptions(clusterName))
		if err != nil {
			return false, err
		}
		if len(pvList.Items) != int(pdCount+tikvCount) {
			return false, errors.Errorf("PV not equal to %d, now: %d", pdCount+tikvCount, len(pvList.Items))
		}

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

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

	tc.Spec.Paused = false

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

	By(fmt.Sprintf("Then Annotations[\"%s\"] is false", label.AnnPaused))
	err = ensureAnnotationEqual("false", ns, clusterName, f)
	Expect(err).NotTo(HaveOccurred())

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

func ensureAnnotationEqual(pausedVal, ns, clusterName string, f *framework.Framework) error {
	return wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		tsList, err := f.Cli.PingcapV1().TidbSets(ns).List(metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				label.New().Cluster(clusterName).Labels(),
			).String(),
		})
		if err != nil {
			return false, nil
		}

		for _, ts := range tsList.Items {
			if p, ok := ts.Annotations[label.AnnPaused]; !ok || p != pausedVal {
				return false, nil
			}
		}

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

		if err != nil {
			return false, nil
		}

		if p, ok := pdSvc.Annotations[label.AnnPaused]; !ok || p != pausedVal {
			return false, nil
		}

		tidbSvc, err := f.KubeCli.CoreV1().Services(ns).Get(fmt.Sprintf("%s-tidb", clusterName), metav1.GetOptions{})
		if err != nil {
			return false, nil
		}

		if p, ok := tidbSvc.Annotations[label.AnnPaused]; !ok || p != pausedVal {
			return false, nil
		}

		return true, nil
	})
}