// 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

package e2e

import (
	"fmt"
	"strings"
	"time"

	yaml "gopkg.in/yaml.v2"

	"github.com/juju/errors"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/scheduler/volume"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	"github.com/pingcap/tidb-operator/test/e2e/bootstrap"
	"github.com/pingcap/tidb-operator/test/e2e/framework"
	"github.com/pingcap/tidb-operator/test/e2e/util"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
)

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

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

	By("Ensure each PV has AnnProvisionerSpaceUsed annotation")
	pvList, err := f.GetAllClusterPVs(clusterName)
	Expect(err).NotTo(HaveOccurred())
	for _, pv := range pvList.Items {
		if pv.Annotations == nil {
			failWrapper(fmt.Sprintf("PV[%s] has empty annotations", pv.Name))
		}
		Expect(pv.Annotations[volumeprovisioner.AnnProvisionerSpaceUsed]).ToNot(Equal(""))
	}

	By("When remove a volume used by a tikv instance from tidb-volumes")
	cm, err := f.KubeCli.CoreV1().ConfigMaps(bootstrap.OperatorNamespace).Get(bootstrap.VolumeConfigMapName, metav1.GetOptions{})
	Expect(err).NotTo(HaveOccurred())

	tv := volume.TiDBVolume{}
	err = yaml.Unmarshal([]byte(cm.Data["volume"]), &tv)
	Expect(err).NotTo(HaveOccurred())

	pvs, err := f.GetAllClusterPVs(clusterName)
	Expect(err).NotTo(HaveOccurred())

	pv := pvs.Items[0]

out:
	for i := 0; i < len(tv.Nodes); i++ {
		node := tv.Nodes[i]
		for j := 0; j < len(node.Directories); j++ {
			dir := node.Directories[j]
			if node.Name == pv.Annotations[volumeprovisioner.AnnProvisionerNode] && dir.Path == pv.Annotations[volumeprovisioner.AnnProvisionerHostPath] {
				util.Logf("removing tidb-volume: %s:%s", node.Name, dir.Path)
				node.Directories = append(node.Directories[:j], node.Directories[j+1:]...)
				util.Logf("node: %s, directories: %+v", node.Name, node.Directories)
				break out
			}
		}
	}

	tvData, err := yaml.Marshal(tv)
	Expect(err).NotTo(HaveOccurred())

	cm.Data = map[string]string{"volume": string(tvData)}
	_, err = f.KubeCli.CoreV1().ConfigMaps(bootstrap.OperatorNamespace).Update(cm)
	Expect(err).NotTo(HaveOccurred())

	By("Then waiting for 1 minute to make sure all tikv instances are running")
	err = wait.Poll(1*time.Second, 1*time.Minute, func() (bool, error) {
		podList, err := f.GetAllTiKVPods(e2eNamespace, clusterName)
		if err != nil {
			return false, nil
		}

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

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

			if !running {
				return false, errors.Errorf("%s/%s is not running", pod.Namespace, pod.Name)
			}
		}

		return false, nil
	})
	if err != nil {
		if !strings.Contains(err.Error(), wait.ErrWaitTimeout.Error()) {
			close(done)
			Expect(err).NotTo(HaveOccurred())
		} else {
			deleteCluster(f, e2eNamespace, clusterName, done, pdCount+tikvCount)
		}
	}
}