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

import (
	"fmt"
	"reflect"
	"sort"
	"strings"
	"time"

	"strconv"

	"github.com/ngaut/log"
	"github.com/pborman/uuid"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/review"
	tsctl "github.com/pingcap/tidb-operator/pkg/tidbset/controller"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/errors"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/pkg/util/namegenerator"
	"github.com/pingcap/tidb-operator/pkg/util/pdutil"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const (
	/* unused
	storeUpState        = "Up"
	storeOfflineState   = "Offline"
	storeDownState      = "Down"
	*/
	storeTombstoneState = "Tombstone"
)

// defaultTiKVRequestSize is tikv default storage size
var defaultTiKVRequestSize = resource.MustParse("50Gi")

const (
	timeout                = 2 * time.Second
	maxGracefulUpgradeTime = 3 * time.Minute
	/* unused
	tikvDownStateName      = "Down"
	*/
)

// Controller is the controller of tidbset for TiKV
type Controller struct {
	cli        versioned.Interface
	kubeCli    kubernetes.Interface
	controller cache.Controller
	store      cache.Store
	queue      workqueue.RateLimitingInterface
}

// NewController creates a new TiKV Controller
func NewController(cli versioned.Interface, kubeCli kubernetes.Interface, resyncDuration time.Duration) *Controller {
	tkc := &Controller{
		cli:     cli,
		kubeCli: kubeCli,
		queue:   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "tikvcontroller"),
	}
	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().TiKV().Labels(),
		).String(),
	}

	tkc.store, tkc.controller = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return tkc.cli.PingcapV1().TidbSets(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return tkc.cli.PingcapV1().TidbSets(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&v1.TidbSet{},
		resyncDuration,
		cache.ResourceEventHandlerFuncs{
			AddFunc: tkc.enqueueTiKV,
			UpdateFunc: func(old, cur interface{}) {
				tkc.enqueueTiKV(cur)
			},
			DeleteFunc: tkc.enqueueTiKV,
		},
	)

	return tkc
}

// Run method runs TiKV Controller
func (tkc *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tkc.queue.ShutDown()
	log.Infof("Starting TiKV controller")
	go tkc.controller.Run(stopCh)

	for i := 0; i < workers; i++ {
		go wait.Until(tkc.worker, time.Second, stopCh)
	}

	<-stopCh
	log.Infof("Shutting down TiKV controller")
}

func (tkc *Controller) worker() {
	for {
		func() {
			key, quit := tkc.queue.Get()
			if quit {
				return
			}
			defer tkc.queue.Done(key)

			if err := tkc.syncTiKVSet(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing TidbSet %v, requeuing: %v", key.(string), err))
				tkc.queue.AddRateLimited(key)
			} else {
				tkc.queue.Forget(key)
			}
		}()
	}
}

func (tkc *Controller) enqueueTiKV(obj interface{}) {
	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	tkc.queue.Add(key)
}

func (tkc *Controller) syncTiKVSet(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing TidbSet[%s] (%v)", key, time.Since(startTime))
	}()

	obj, exists, err := tkc.store.GetByKey(key)
	if err != nil {
		return errors.Trace(err)
	}
	if !exists {
		return tkc.deleteTiKVSet(key)
	}

	ts, ok := obj.(*v1.TidbSet)
	if !ok {
		return errors.Trace(fmt.Errorf("object %v is not a *TidbSet", obj))
	}

	if ts.Paused() {
		log.Infof("TidbSet[%s/%s] controller paused, skip syncing", ts.GetNamespace(), ts.GetName())
		return nil
	}

	ts, pods, err := tkc.syncReplicasAndStatusFor(ts)
	if err != nil {
		return errors.Trace(err)
	}

	return tkc.syncTiKVPodFor(ts, pods)
}

func (tkc *Controller) syncReplicasAndStatusFor(ts *v1.TidbSet) (*v1.TidbSet, []corev1.Pod, error) {
	ns := ts.GetNamespace()
	name := ts.GetName()
	var replicas int32
	var normalReplicas int32
	var readyReplicas int32
	listOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(ts.Spec.Selector.MatchLabels).String(),
	}
	clusterName := strings.TrimSuffix(ts.OwnerReferences[0].Name, "-tikvset")

	podList, listErr := tkc.kubeCli.CoreV1().Pods(ns).List(listOptions)
	if listErr != nil {
		return ts, nil, errors.Trace(listErr)
	}

	if len(podList.Items) == 0 {
		log.Infof("TidbSet: [%s/%s]'s pods count is zero, clearing its status", ns, name)
		ts.Status = v1.TidbSetStatus{}
		var err error
		ts, err = tkc.cli.PingcapV1().TidbSets(ns).Update(ts)
		if err != nil {
			errDetail := errors.Logf(err, "failed to update TidbSet: [%s/%s] status", ns, name)
			return ts, podList.Items, errDetail
		}

		log.Infof("update TidbSet: [%s/%s] status successfully", ns, name)
		return ts, podList.Items, nil
	}

	ts, syncErr := tkc.syncTiKVStatusFor(ts)
	if syncErr != nil {
		log.Errorf("failed to sync TiKV status for TidbSet: [%s/%s], %v", ns, name, syncErr)
		return ts, podList.Items, errors.Trace(syncErr)
	}

	pdCli := util.NewPDClientForCluster(ns, clusterName, timeout)
	config, confErr := pdCli.GetConfig()
	if confErr != nil {
		log.Errorf("failed to get PD config for TidbCluster: [%s/%s], %v", ns, clusterName, confErr)
		return ts, podList.Items, errors.Trace(confErr)
	}
	if config == nil {
		return ts, podList.Items, errors.Errorf("pdCli.GetConfig returns nil")
	}

	for _, pod := range podList.Items {
		replicas++
		podName := pod.GetName()
		if pod.Annotations == nil {
			pod.Annotations = make(map[string]string)
		}
		ann := pod.Annotations[label.AnnShouldDeletedTiKV]
		if ann == podName {
			log.Infof("Pod: [%s/%s]'s annotation[%s] is: [%s], skipping", ns, podName, label.AnnShouldDeletedTiKV, ann)
			continue
		}

		ps, ok := ts.Status.TiKVStatus[podName]
		if ok {
			var dur time.Duration
			if !ps.LastHeartbeatTime.IsZero() {
				dur = time.Since(ps.LastHeartbeatTime.Time)
			}
			// Create a new TiKV instance If TiKV unreachable longer than PD's MaxStoreDownTime
			if dur.Seconds() > config.Schedule.MaxStoreDownTime.Seconds() {
				log.Infof("pod: [%s/%s]'s LastHeartbeatTS: [%s] is greater than: [%s]",
					ns, podName, ps.LastHeartbeatTime.String(), config.Schedule.MaxStoreDownTime.String())
				continue
			}
		}

		normalReplicas++
		if ts.IsReady(&pod) {
			readyReplicas++
		}
	}

	ts.Status.Replicas = replicas
	ts.Status.NormalReplicas = normalReplicas
	ts.Status.ReadyReplicas = readyReplicas
	ts, err := tkc.cli.PingcapV1().TidbSets(ns).Update(ts)
	if err != nil {
		log.Errorf("failed to update TidbSet: [%s/%s] status, %v", ns, name, err)
		return ts, podList.Items, errors.Trace(err)
	}

	log.Infof("update TidbSet: [%s/%s] status successfully", ns, name)
	return ts, podList.Items, nil
}

func (tkc *Controller) syncTiKVStatusFor(ts *v1.TidbSet) (*v1.TidbSet, error) {
	ns := ts.GetNamespace()
	tsName := ts.GetName()
	clusterName := ts.OwnerReferences[0].Name

	pdCli := util.NewPDClientForCluster(ns, clusterName, timeout)
	// This only returns Up/Down/Offline stores
	storesInfo, err := pdCli.GetStores()
	if err != nil {
		log.Errorf("failed to get stores from PD for TidbCluster: [%s/%s], %v", ns, clusterName, err)
		return ts, errors.Trace(err)
	}

	tombstoneStoresInfo, err := pdCli.GetTombStoneStores()
	if err != nil {
		log.Errorf("failed to get tombstone stores from PD for TidbCluster: [%s/%s], %v", ns, clusterName, err)
		return ts, errors.Trace(err)
	}

	stores := append(storesInfo.Stores, tombstoneStoresInfo.Stores...)

	if len(stores) == 0 {
		log.Debugf("pdCli.GetStores and pdCli.GetTombStoneStores both return nil in [%s/%s]", ns, clusterName)
		return ts, nil
	}

	listOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(ts.Spec.Selector.MatchLabels).String(),
	}
	podList, err := tkc.kubeCli.CoreV1().Pods(ns).List(listOptions)
	if err != nil {
		return ts, errors.Trace(err)
	}

	tikvStatus := make(map[string]v1.TiKVStatus)

	for _, store := range stores {
		var podName string
		if store.Store == nil || store.Status == nil {
			continue
		}

		storeID := fmt.Sprintf("%d", store.Store.GetId())
		ip := strings.Split(store.Store.GetAddress(), ":")[0]
		status := v1.TiKVStatus{
			ID:                storeID,
			IP:                ip,
			State:             store.Store.StateName,
			LastHeartbeatTime: metav1.Time{Time: store.Status.LastHeartbeatTS},
		}

		var tikvPod *corev1.Pod
		for _, pod := range podList.Items {
			pn := pod.GetName()
			if _, ok := tikvStatus[pn]; ok {
				continue
			}

			if pod.Spec.Hostname != "" && pod.Spec.Subdomain != "" {
				domainName := fmt.Sprintf("%s.%s.%s.svc", pod.Spec.Hostname, pod.Spec.Subdomain, pod.GetNamespace())
				if domainName == ip {
					tikvPod = &pod
					break
				}
			} else {
				// for backward compatibility, old operator uses pod ip in tikvset
				podIP := pod.Status.PodIP
				if podIP == ip {
					tikvPod = &pod
					break
				}
			}
		}

		if tikvPod == nil {
			continue
		}

		podName = tikvPod.GetName()
		err = tkc.setStoreLabelsForPod(pdCli, store, tikvPod)
		if err != nil {
			log.Errorf("failed to setStoreLabelsForPod: [%s/%s], %v", ns, podName, err)
			return ts, errors.Trace(err)
		}

		// avoid LastHeartbeatTime be overwrite by zero time when pd lost LastHeartbeatTime
		if status.LastHeartbeatTime.IsZero() {
			if oldStatus, ok := ts.Status.TiKVStatus[podName]; ok {
				log.Warningf("the pod:%s's store LastHeartbeatTime is zero,so will keep in %v", podName, oldStatus.LastHeartbeatTime)
				status.LastHeartbeatTime = oldStatus.LastHeartbeatTime
			}
		}

		tikvStatus[podName] = status
	}

	if !reflect.DeepEqual(tikvStatus, ts.Status.TiKVStatus) {
		ts.Status.TiKVStatus = tikvStatus
		ts, err = tkc.cli.PingcapV1().TidbSets(ts.Namespace).Update(ts)
		if err != nil {
			log.Errorf("failed to update TidbSet: [%s/%s], %v", ns, tsName, err)
			return ts, errors.Trace(err)
		}

		log.Infof("update TidbSet: [%s/%s] status successfully.", ns, tsName)
	}

	return ts, nil
}

func (tkc *Controller) deleteTiKVSet(key string) error {
	ns, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return errors.Trace(err)
	}

	_, err = tkc.cli.PingcapV1().TidbSets(ns).Get(name, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			log.Infof("TidbSet: [%s/%s] was been deleted, deleting all related tikv pods", ns, name)
			return tkc.deleteTiKVPods(ns, name)
		}

		return err
	}

	log.Warnf("TidbSet: [%s/%s] exists, skipping...", ns, name)
	return nil
}

func (tkc *Controller) createPVCForTiKVPod(tidbset *v1.TidbSet, podName string) (*corev1.PersistentVolumeClaim, error) {
	ns := tidbset.Namespace
	pvcName := fmt.Sprintf("pvc-%s-%s", tidbset.Name, uuid.New())
	storage := defaultTiKVRequestSize

	if tidbset.Annotations != nil {
		storageFromAnnotation, ok := tidbset.Annotations[v1.AnnotationStorageSize]
		if ok {
			q, err := resource.ParseQuantity(storageFromAnnotation)
			if err != nil {
				return nil, errors.Errorf(
					"can't parse storage from: [%s] when creating PVC: %s for TiKVSet: [%s/%s], %v",
					storageFromAnnotation,
					pvcName,
					ns,
					tidbset.Name,
					err,
				)
			}
			storage = q
		}
	}

	clusterName := tidbset.OwnerReferences[0].Name
	storageClassName := volumeprovisioner.StorageClassName
	newPVC := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      pvcName,
			Namespace: ns,
			Labels:    label.New().Cluster(clusterName).TiKV(),
			Annotations: map[string]string{
				label.AnnPodNameKey: podName,
			},
		},
		Spec: corev1.PersistentVolumeClaimSpec{
			AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
			Resources: corev1.ResourceRequirements{
				Requests: corev1.ResourceList{corev1.ResourceStorage: storage},
			},
			StorageClassName: &storageClassName,
		},
	}

	pvc, err := tkc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Create(newPVC)
	if err != nil {
		log.Errorf("PVC[%s/%s] created failed: %v", ns, pvcName, err)
		return nil, errors.Trace(err)
	}
	log.Infof("PVC[%s/%s] created successfully", ns, pvcName)
	return pvc, nil
}

func (tkc *Controller) createTiKVInstanceFor(tidbset *v1.TidbSet) error {
	clusterName := tidbset.OwnerReferences[0].Name
	podName := namegenerator.SimpleNameGenerator.GenerateName(fmt.Sprintf("%s-tikv-", clusterName))
	pvc, err := tkc.createPVCForTiKVPod(tidbset, podName)
	if err != nil {
		return errors.Trace(err)
	}

	uid := tidbset.UID
	ns := tidbset.Namespace
	controller := true

	ownerRef := metav1.OwnerReference{
		APIVersion: tsctl.ControllerKind.GroupVersion().String(),
		Kind:       tsctl.ControllerKind.Kind,
		Name:       tidbset.Name,
		UID:        uid,
		Controller: &controller,
	}

	podSpec := tidbset.Spec.Template.Spec.DeepCopy()
	podSpec.Hostname = podName

	for _, vol := range podSpec.Volumes {
		if vol.Name == v1.TiDBVolumeName && vol.VolumeSource.PersistentVolumeClaim != nil {
			vol.VolumeSource.PersistentVolumeClaim.ClaimName = pvc.Name
		}
	}

	podSpec.SchedulerName = util.TiDBScheduler

	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:            podName,
			Namespace:       ns,
			Labels:          tidbset.Spec.Selector.MatchLabels,
			OwnerReferences: []metav1.OwnerReference{ownerRef},
			Annotations:     util.AnnProm(nil, "/metrics"),
		},
		Spec: *podSpec,
	}
	_, err = tkc.kubeCli.CoreV1().Pods(ns).Create(pod)
	if err != nil {
		log.Errorf("failed to create Pod: [%s/%s], %v", ns, podName, err)
		return errors.Trace(err)
	}

	log.Infof("create Pod: [%s/%s] successfully", ns, podName)
	return nil
}

func (tkc *Controller) deleteTiKVPodFor(ts *v1.TidbSet, pod *corev1.Pod) error {
	ns := ts.GetNamespace()
	clusterName := ts.OwnerReferences[0].Name
	podName := pod.GetName()
	storeID := ts.Status.TiKVStatus[podName].ID
	storeState := ts.Status.TiKVStatus[podName].State
	if len(storeID) == 0 {
		return errors.Errorf("TidbSet: [%s/%s]'s Pod: [%s/%s] storeID is empty", ns, ts.GetName(), ns, podName)
	}

	err := util.NewPDClientForCluster(ns, clusterName, timeout).DeleteStore(storeID)
	if err != nil {
		log.Errorf("failed to delete Pod: [%s/%s]'s store ID: [%s] from cluster", ns, podName, storeID)
		return errors.Trace(err)
	}

	log.Infof("delete Pod: [%s/%s]'s store ID: [%s] from cluster successfully", ns, podName, storeID)

	if storeState == storeTombstoneState {
		err := tkc.markPVTombstoneAndReleasePVCFor(pod)
		if err != nil {
			return errors.Trace(err)
		}

		err = tkc.kubeCli.CoreV1().Pods(ns).Delete(podName, nil)
		if err != nil {
			log.Errorf("failed to delete Pod: [%s/%s], %v", ns, podName, err)
			return errors.Trace(err)
		}

		log.Infof("delete Pod: [%s/%s] successfully", ns, podName)
	} else {
		log.Warnf("Pod: [%s/%s]'s state is not: [%s], skipping", ns, podName, storeTombstoneState)
	}

	return nil
}

func (tkc *Controller) setPodShouldDeleted(pod *corev1.Pod) error {
	ns := pod.GetNamespace()
	podName := pod.GetName()
	if pod.Annotations == nil {
		pod.Annotations = make(map[string]string)
	}
	if pod.Annotations[label.AnnShouldDeletedTiKV] != podName {
		pod.Annotations[label.AnnShouldDeletedTiKV] = podName
		_, err := tkc.kubeCli.CoreV1().Pods(ns).Update(pod)
		if err != nil {
			log.Errorf("failed to update Pod: [%s/%s]'s annotations: [%s] to: [%s], %v", ns, podName, label.AnnShouldDeletedTiKV, podName, err)
			return errors.Trace(err)
		}

		log.Infof("update Pod: [%s/%s]'s annotations: [%s] to: %s successfully", ns, podName, label.AnnShouldDeletedTiKV, podName)
	}

	return nil
}

func (tkc *Controller) syncTiKVPodFor(ts *v1.TidbSet, pods []corev1.Pod) error {
	ns := ts.GetNamespace()
	clusterName := ts.OwnerReferences[0].Name
	tsName := ts.GetName()

	// Check if PD is started, TiKV can only be started after PD is started
	pdListOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(label.New().Cluster(clusterName).PD().Labels()).String(),
	}
	pdPodList, listErr := tkc.kubeCli.CoreV1().Pods(ns).List(pdListOptions)
	if listErr != nil {
		return errors.Trace(listErr)
	}

	pdsetName := fmt.Sprintf("%s-pdset", clusterName)
	pdset, setErr := tkc.cli.PingcapV1().TidbSets(ns).Get(pdsetName, metav1.GetOptions{})
	if setErr != nil {
		return errors.Trace(setErr)
	}

	var pdStartedCount int32
	for _, pod := range pdPodList.Items {
		if pdset.IsReady(&pod) {
			pdStartedCount++
		}
	}

	if pdStartedCount < pdset.Spec.Replicas {
		log.Warnf("At least [%d] PDs must been started, now: [%d] in [%s/%s]",
			pdset.Spec.Replicas, pdStartedCount, ns, clusterName)
		return nil
	}

	for _, pod := range pods {
		podName := pod.GetName()
		if pod.Annotations[label.AnnShouldDeletedTiKV] == podName {
			log.Infof("Pod: [%s/%s] have annotations: [%s]", ns, podName, label.AnnShouldDeletedTiKV)
			if ts.Status.NormalReplicas < 3 && ts.Spec.Replicas == 3 {
				log.Infof("Add a new one before deleting when there is only 3 TiKV pods, ts.Status.NormalReplicas: [%d] in [%s/%s]",
					ts.Status.NormalReplicas, ns, clusterName)
				return tkc.createTiKVInstanceFor(ts)
			}

			msg := fmt.Sprintf("Pod: [%s/%s] was marked should deleted, deleting it", ns, podName)
			log.Info(msg)

			err := review.Review(tkc.cli, review.Info{
				UID:         string(pod.UID),
				Namespace:   ns,
				ClusterName: clusterName,
				Reason:      "ShouldDeleteTiKV",
				Message:     msg,
			})
			if err != nil {
				return errors.Trace(err)
			}

			return tkc.deleteTiKVPodFor(ts, &pod)
		}
	}

	for _, pod := range pods {
		podName := pod.GetName()
		if status, found := ts.Status.TiKVStatus[podName]; found && status.State == storeTombstoneState {
			msg := fmt.Sprintf("TidbCluster: [%s/%s]'s Pod: [%s/%s]'s State is tombstone, deleting", ns, clusterName, ns, podName)
			log.Info(msg)

			err := review.Review(tkc.cli, review.Info{
				UID:         string(pod.UID),
				Namespace:   ns,
				ClusterName: clusterName,
				Reason:      "DeleteTombstoneTiKV",
				Message:     msg,
			})
			if err != nil {
				return errors.Trace(err)
			}

			err = tkc.setPodShouldDeleted(&pod)
			if err != nil {
				return errors.Trace(err)
			}

			err = tkc.deleteTiKVPodFor(ts, &pod)
			if err != nil {
				return errors.Trace(err)
			}
		}
	}

	if ts.Status.NormalReplicas < ts.Spec.Replicas {
		log.Infof("TidbSet: [%s/%s] Status.NormalReplicas = %d Spec.Replicas = %d", ns, tsName, ts.Status.NormalReplicas, ts.Spec.Replicas)
		for i := 0; i < int(ts.Spec.Replicas-ts.Status.NormalReplicas); i++ {
			err := tkc.createTiKVInstanceFor(ts)
			if err != nil {
				return errors.Trace(err)
			}
		}

		return nil
	}

	if ts.Status.NormalReplicas > ts.Spec.Replicas {
		log.Infof("TidbSet: [%s/%s]'s Status.NormalReplicas is: [%d], Spec.Replicas is: [%d], decreasing", ns, tsName, ts.Status.NormalReplicas, ts.Spec.Replicas)
		sort.Sort(tsctl.ActivePods(pods))

		var p *corev1.Pod
		if len(pods) > 0 {
			p = &pods[0]
		}

		if p == nil {
			return errors.Errorf("ts.Status.Replicas: [%d], ts.Status.NormalReplicas: [%d], ts.Spec.Replicas: [%d], but pods count is zero", ts.Status.Replicas, ts.Status.NormalReplicas, ts.Spec.Replicas)
		}

		msg := fmt.Sprintf("TidbSet: [%s/%s]'s NormalReplicas: [%d] is greater than Replicas: [%d], deleting one pod: [%s/%s]",
			ns, ts.GetName(), ts.Status.NormalReplicas, ts.Spec.Replicas, ns, p.GetName())
		log.Info(msg)

		err := review.Review(tkc.cli, review.Info{
			UID:         string(p.UID),
			Namespace:   ns,
			ClusterName: clusterName,
			Reason:      "DecreaseTiKVReplicas",
			Message:     msg,
		})
		if err != nil {
			return errors.Trace(err)
		}

		if err := tkc.setPodShouldDeleted(p); err != nil {
			return errors.Trace(err)
		}

		return tkc.deleteTiKVPodFor(ts, p)
	}

	//upgrade pod
	pdCli := util.NewPDClientForCluster(ns, clusterName, timeout)
	upgradingPod, err := tkc.markTikvPodToUpgrade(ts, pdCli, pods)
	if err != nil {
		return errors.Trace(err)
	}
	if upgradingPod != nil {
		if err := tkc.gracefulUpgradePod(ts, pdCli, upgradingPod); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

//graceful upgrade the pod.
func (tkc *Controller) gracefulUpgradePod(ts *v1.TidbSet, pdCli *pdutil.Client, pod *corev1.Pod) error {
	newImage := tsctl.GetImageByPodSpec(v1.TiKVContainerType, &ts.Spec.Template.Spec)
	oldImage := tsctl.GetImageByPodSpec(v1.TiKVContainerType, &pod.Spec)
	tsName := ts.GetName()
	ns := ts.GetNamespace()
	podName := pod.GetName()

	if oldImage == newImage {
		return nil
	}

	if _, ok := pod.Annotations[label.AnnGracefulUpgradeTiKVStartTime]; !ok {
		log.Errorf("the pod: [%s/%s]'s Annotations[AnnGracefulUpgradeTiKVStartTime] is not exist.", ns, podName)
		return nil
	}
	startTimeStr := pod.Annotations[label.AnnGracefulUpgradeTiKVStartTime]
	startTime, parseErr := time.Parse(time.RFC3339, startTimeStr)
	if parseErr != nil {
		return errors.Trace(parseErr)
	}

	podStatus, exist := ts.Status.TiKVStatus[podName]
	if !exist {
		return errors.Errorf("TidbSet: [%s/%s]'s Pod: [%s] store status is nil", ns, tsName, podName)
	}

	storeID, parseIntErr := strconv.ParseUint(podStatus.ID, 10, 0)
	if parseIntErr != nil {
		return errors.Logf(parseIntErr, "TidbSet: [%s/%s]'s Pod: [%s] storeID parse to unit64 failed", ns, tsName, podName)
	}

	storeInfo, storeErr := pdCli.GetStore(storeID)
	if storeErr != nil {
		return errors.Logf(storeErr, "TidbSet: [%s/%s]'s Pod: [%s] get store:[%d] failed", ns, tsName, podName, storeID)
	}

	now := time.Now()
	if storeInfo.Status.LeaderCount == 0 || now.After(startTime.Add(maxGracefulUpgradeTime)) {
		log.Infof("starting to upgrade TidbSet:[%s/%s]'s Pod: [%s],when the leaderCount is %d and %d seconds elapsed", ns, tsName, podName, storeInfo.Status.LeaderCount, now.Sub(startTime).Seconds())
		err := wait.Poll(3*time.Second, 30*time.Second, func() (bool, error) {
			updatePod, err := tkc.kubeCli.CoreV1().Pods(ns).Get(podName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			updatePodName := updatePod.GetName()
			delete(updatePod.Annotations, label.AnnGracefulUpgradeTiKVStartTime)
			tsctl.SetImageByPodSpec(v1.TiKVContainerType, &updatePod.Spec, newImage)
			if _, err := tkc.kubeCli.CoreV1().Pods(updatePod.Namespace).Update(updatePod); err != nil {
				log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] upgrade to image: [%s] failed: %v", ns, tsName, updatePodName, newImage, err)
				return false, nil
			}
			if err := pdCli.EndEvictLeader(podStatus.ID); err != nil {
				log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] end evict leader failed: %v", ns, tsName, updatePodName, err)
				return false, nil
			}
			log.Infof("TidbSet: [%s/%s]'s Pod: [%s] upgrade to image: [%s] successfully", ns, tsName, updatePodName, newImage)
			return true, nil
		})
		if err != nil {
			return errors.Trace(err)
		}
	} else {
		log.Infof("TidbSet: [%s/%s]'s Pod: [%s] is evicting leader,the leader count is %d", ns, tsName, pod.Name, storeInfo.Status.LeaderCount)
		return nil
	}
	return nil
}

func (tkc *Controller) markTikvPodToUpgrade(ts *v1.TidbSet, pdCli *pdutil.Client, pods []corev1.Pod) (*corev1.Pod, error) {
	newImage := tsctl.GetImageByPodSpec(v1.TiKVContainerType, &ts.Spec.Template.Spec)
	ns := ts.GetNamespace()
	tsName := ts.GetName()

	var markPod *corev1.Pod
	var markStoreID uint64
	for i, pod := range pods {
		oldImage := tsctl.GetImageByPodSpec(v1.TiKVContainerType, &pod.Spec)
		if oldImage != newImage {
			if _, ok := pod.Annotations[label.AnnGracefulUpgradeTiKVStartTime]; ok {
				return &pod, nil
			}

			podName := pod.GetName()
			podStatus, exist := ts.Status.TiKVStatus[podName]
			if !exist {
				log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] store status is nil", ns, tsName, podName)
				continue
			}

			storeID, err := strconv.ParseUint(podStatus.ID, 10, 0)
			if err != nil {
				log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] parse failed:%v", ns, tsName, podName, err)
				continue
			}

			if markPod == nil {
				markPod = &pods[i]
				markStoreID = storeID
			}
		}
	}

	if markPod == nil {
		return nil, nil
	}

	markPodName := markPod.GetName()
	if ts.Status.ReadyReplicas != ts.Status.Replicas {
		log.Infof("TidbSet: [%s/%s]'s ReadyReplicas does not equal ts.Status.Replicas,pod: [%s] can not be marked to upgrade ", ns, tsName, markPodName)
		return nil, nil
	}

	err := wait.Poll(3*time.Second, 30*time.Second, func() (bool, error) {
		err := pdCli.BeginEvictLeader(markStoreID)
		if err != nil {
			log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] begin evict leader failed: %v", ns, tsName, markPodName, err)
			return false, nil
		}

		markPod, err = tkc.kubeCli.CoreV1().Pods(ns).Get(markPodName, metav1.GetOptions{})
		if err != nil {
			log.Errorf("Get TidbSet: [%s/%s]'s Pod: [%s] failed before update : %v", ns, tsName, markPodName, err)
			return false, nil
		}

		markPod.Annotations[label.AnnGracefulUpgradeTiKVStartTime] = time.Now().Format(time.RFC3339)
		_, err = tkc.kubeCli.CoreV1().Pods(ns).Update(markPod)
		if err != nil {
			log.Errorf("TidbSet: [%s/%s]'s Pod: [%s] set AnnGracefulUpgradeTikvStartTime failed : %v", ns, tsName, markPodName, err)
			return false, nil
		}

		log.Infof("TidbSet: [%s/%s]'s Pod: [%s] is set the next upgrading pod.", ns, tsName, markPodName)
		return true, nil
	})

	if err != nil {
		return nil, err
	}

	return markPod, nil

}

func (tkc *Controller) deleteTiKVPods(ns, name string) error {
	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(strings.TrimSuffix(name, "-tikvset")).TiKV().Labels(),
		).String(),
	}

	return wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
		podList, err := tkc.kubeCli.CoreV1().Pods(ns).List(opts)
		if err != nil {
			return false, nil
		}

		for _, pod := range podList.Items {
			podName := pod.GetName()
			err := tkc.releasePVCFor(&pod)
			if err != nil {
				log.Errorf("failed to releasePVCFor Pod[%s/%s], %v", ns, podName, err)
				return false, nil
			}

			err = tkc.kubeCli.CoreV1().Pods(ns).Delete(podName, nil)
			if err != nil {
				log.Errorf("failed to delete Pod[%s/%s], %v", ns, podName, err)
				return false, nil
			}
			log.Infof("Pod: [%s/%s] deleted successfully.", ns, podName)
		}

		return true, nil
	})
}

func (tkc *Controller) markPVTombstoneAndReleasePVCFor(pod *corev1.Pod) error {
	ns := pod.GetNamespace()
	podName := pod.GetName()

	for _, vol := range pod.Spec.Volumes {
		if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
			claimName := vol.PersistentVolumeClaim.ClaimName
			pvc, err := tkc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Get(claimName, metav1.GetOptions{})
			if err != nil {
				return errors.Trace(err)
			}

			pvName := pvc.Spec.VolumeName
			pv, err := tkc.kubeCli.CoreV1().PersistentVolumes().Get(pvName, metav1.GetOptions{})
			if err != nil {
				return errors.Trace(err)
			}

			now := time.Now().Format(time.RFC3339)
			pv.Annotations[volumeprovisioner.AnnPVTombstoned] = now
			_, err = tkc.kubeCli.CoreV1().PersistentVolumes().Update(pv)
			if err != nil {
				log.Errorf("failed to set PV: %s annotation: %s, %v", pvName, volumeprovisioner.AnnPVTombstoned, err)
				return errors.Trace(err)
			}
			log.Infof("Set PV: %s annotation: %s to %s successfully", pvName, volumeprovisioner.AnnPVTombstoned, now)

			err = tkc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(claimName, nil)
			if err != nil && !apierrors.IsNotFound(err) {
				log.Errorf("Pod: [%s/%s]'s PVC: [%s/%s] released failed: %v", ns, podName, ns, claimName, err)
				return errors.Trace(err)
			}
			log.Infof("Pod: [%s/%s]'s PVC[%s/%s] released successfully.", ns, podName, ns, claimName)
			return nil
		}
	}

	return errors.Errorf("can't find PVC from Pod: [%s/%s]", ns, podName)
}

func (tkc *Controller) releasePVCFor(pod *corev1.Pod) error {
	ns := pod.GetNamespace()
	podName := pod.GetName()

	for _, vol := range pod.Spec.Volumes {
		if vol.Name == v1.TiDBVolumeName && vol.PersistentVolumeClaim != nil {
			claimName := vol.PersistentVolumeClaim.ClaimName
			err := tkc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(claimName, nil)
			if err != nil && !apierrors.IsNotFound(err) {
				log.Errorf("Pod: [%s/%s]'s PVC: [%s/%s] released failed: %v", ns, podName, ns, claimName, err)
				return errors.Trace(err)
			}
			log.Infof("Pod: [%s/%s]'s PVC[%s/%s] released successfully.", ns, podName, ns, claimName)
			return nil
		}
	}

	return errors.Errorf("can't find PVC from Pod: [%s/%s]", ns, podName)
}