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

func testDecrease(f *framework.Framework) {
	clusterName := "e2e-decrease"
	var pdCount int32 = 5
	var tidbCount int32 = 1
	var tikvCount int32 = 5

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

	var newTiKVCount int32 = 3
	var newPDCount int32 = 3
	By(fmt.Sprintf("When resize tikv's replicas to %d and pd's replicas to %d", newTiKVCount, newPDCount))
	tc, err := f.Cli.PingcapV1().TidbClusters(e2eNamespace).Get(clusterName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

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

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

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

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

		pdPodList, err := f.GetAllPDPods(e2eNamespace, clusterName)
		if err != nil {
			return false, nil
		}

		if len(pdPodList.Items) != int(newPDCount) {
			return false, nil
		}

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

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

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

	By("And tikv plus pd's pv size is still 10 and 4 of them are tombstoned")
	err = wait.Poll(5*time.Second, 30*time.Minute, func() (bool, error) {
		pvList, err := f.KubeCli.CoreV1().PersistentVolumes().List(label.ClusterListOptions(clusterName))
		if err != nil {
			return false, nil
		}

		if len(pvList.Items) != 10 {
			return true, errors.New("PV size is not 10")
		}

		tombstoneCount := 0
		for _, pv := range pvList.Items {
			if pv.Annotations[volumeprovisioner.AnnPVTombstoned] != "" {
				tombstoneCount++
			}
		}

		if tombstoneCount == 4 {
			return true, nil
		}

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

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

type vol struct {
	node     string
	hostPath string
}

func getVolume(f *framework.Framework, pod *corev1.Pod) (*corev1.PersistentVolume, *corev1.PersistentVolumeClaim, error) {
	var pvcName string
	for _, vol := range pod.Spec.Volumes {
		if vol.Name == v1.TiDBVolumeName && vol.VolumeSource.PersistentVolumeClaim != nil {
			pvcName = vol.VolumeSource.PersistentVolumeClaim.ClaimName
		}
	}
	if pvcName == "" {
		return nil, nil, errors.New("pvc not found")
	}

	pvc, err := f.KubeCli.CoreV1().PersistentVolumeClaims(e2eNamespace).Get(pvcName, metav1.GetOptions{})
	if err != nil {
		return nil, nil, err
	}

	pv, err := f.KubeCli.CoreV1().PersistentVolumes().Get(pvc.Spec.VolumeName, metav1.GetOptions{})
	if err != nil {
		return nil, nil, err
	}

	return pv, pvc, nil
}