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

	"fmt"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	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 testIncrease(f *framework.Framework) {
	clusterName := "e2e-increase"
	var pdCount int32 = 3
	var tidbCount int32 = 1
	var tikvCount int32 = 3

	var newPDCount int32 = 5
	var newTiDBCount int32 = 3
	var newTiKVCount int32 = 5

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

	By("When resize pd's replicas to 5, tidb's replicas to 3 and tikv's replicas to 5")
	tc, err := f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tc.Spec.PD.Size = newPDCount
	tc.Spec.TiDB.Size = newTiDBCount
	tc.Spec.TiKV.Size = newTiKVCount

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

	By(fmt.Sprintf("Then pd's pods size are %d", newPDCount))
	err = wait.Poll(5*time.Second, 30*time.Minute, func() (bool, error) {
		podList, err := f.GetAllPDPods(e2eNamespace, clusterName)
		if err != nil {
			return false, nil
		}

		if len(podList.Items) != 5 {
			return false, nil
		}

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

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

	By(fmt.Sprintf("And tidb's pods size are %d", newTiDBCount))
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		podList, err := f.GetAllTiDBPods(e2eNamespace, clusterName)
		if err != nil {
			return false, nil
		}

		if len(podList.Items) != 3 {
			return false, nil
		}

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

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

	By(fmt.Sprintf("And tikv's pods size are %d", newTiKVCount))
	err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		podList, err := f.GetAllTiKVPods(e2eNamespace, clusterName)
		if err != nil {
			return false, nil
		}

		if len(podList.Items) != 5 {
			return false, nil
		}

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

	By("When we kill all pd and tikv pods")
	volsOld := make(map[string]struct{})
	pvcList, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).List(label.ClusterListOptions(clusterName))
	Expect(err).NotTo(HaveOccurred())

	for _, pvc := range pvcList.Items {
		node := pvc.Annotations[volumeprovisioner.AnnProvisionerNode]
		hostPath := pvc.Annotations[volumeprovisioner.AnnProvisionerHostPath]
		volsOld[fmt.Sprintf("%s-%s", node, hostPath)] = struct{}{}
	}

	delPVCOpts := label.ClusterListOptions(clusterName)
	err = f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).DeleteCollection(nil, delPVCOpts)
	Expect(err).NotTo(HaveOccurred())
	delPodOpts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).Labels(),
		).String(),
	}
	err = f.KubeCli.CoreV1().Pods(e2eNamespace).DeleteCollection(nil, delPodOpts)
	Expect(err).NotTo(HaveOccurred())

	By("Then new created pods will use the previous PV")
	allResourcesIsReady(f, e2eNamespace, clusterName, newPDCount, newTiDBCount, newTiKVCount)

	volsNew := make(map[string]struct{})
	pvcList, err = f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).List(label.ClusterListOptions(clusterName))
	Expect(err).NotTo(HaveOccurred())

	for _, pvc := range pvcList.Items {
		node := pvc.Annotations[volumeprovisioner.AnnProvisionerNode]
		hostPath := pvc.Annotations[volumeprovisioner.AnnProvisionerHostPath]
		volsNew[fmt.Sprintf("%s-%s", node, hostPath)] = struct{}{}
	}

	Expect(volsNew).To(Equal(volsOld))

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

	By("When delete the TidbCluster")
	err = f.Cli.PingcapV1().TidbClusters(e2eNamespace).Delete(clusterName, nil)
	Expect(err).NotTo(HaveOccurred())

	By(fmt.Sprintf("And all its PVs will have Annotations[%s] set", volumeprovisioner.AnnPVClusterDeleted))
	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 {
			if pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] == "" {
				return false, nil
			}
		}

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

	By("When manual delete a PV")
	pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(label.ClusterListOptions(clusterName))
	Expect(err).NotTo(HaveOccurred())
	pv := pvList.Items[0]
	pvName := pv.GetName()
	pv.Annotations[volumeprovisioner.AnnPVManualDeleted] = time.Now().Format(time.RFC3339)
	_, err = f.KubeCli.CoreV1().PersistentVolumes().Update(&pv)
	Expect(err).NotTo(HaveOccurred())

	By("Then this PV should be deleted")
	err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
		_, err := f.KubeCli.CoreV1().PersistentVolumes().Get(pvName, metav1.GetOptions{})
		if err != nil && apierrs.IsNotFound(err) {
			return true, nil
		}

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

	deleteCluster(f, e2eNamespace, clusterName, done, newPDCount+newTiKVCount-1)
}