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

import (
	"fmt"
	"net/url"
	"reflect"
	"sort"
	"strings"
	"time"

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

var (
	keyFunc              = cache.DeletionHandlingMetaNamespaceKeyFunc
	defaultPDRequestSize = resource.MustParse("10Gi")
)

const (
	timeout = 2 * time.Second
)

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

// NewController creates a new PD Controller
func NewController(cli versioned.Interface, kubeCli kubernetes.Interface, resyncDuration time.Duration) *Controller {
	pc := &Controller{
		cli:     cli,
		kubeCli: kubeCli,
		queue:   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pdcontroller"),
	}
	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().PD().Labels(),
		).String(),
	}
	pc.store, pc.controller = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(_ metav1.ListOptions) (runtime.Object, error) {
				return pc.cli.PingcapV1().TidbSets(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(_ metav1.ListOptions) (watch.Interface, error) {
				return pc.cli.PingcapV1().TidbSets(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&v1.TidbSet{},
		resyncDuration,
		cache.ResourceEventHandlerFuncs{
			AddFunc: pc.enqueuePD,
			UpdateFunc: func(old, cur interface{}) {
				pc.enqueuePD(cur)
			},
			DeleteFunc: pc.enqueuePD,
		})
	return pc
}

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

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

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

func (pc *Controller) enqueuePD(obj interface{}) {
	key, err := keyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}
	pc.queue.Add(key)
}

func (pc *Controller) worker() {
	for {
		func() {
			key, quit := pc.queue.Get()
			if quit {
				return
			}
			defer pc.queue.Done(key)
			if err := pc.syncPDSet(key.(string)); err != nil {
				utilruntime.HandleError(fmt.Errorf("Error syncing TidbSet %v, requeuing: %v", key.(string), err))
				pc.queue.AddRateLimited(key)
			} else {
				pc.queue.Forget(key)
			}
		}()
	}
}

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

	obj, exists, err := pc.store.GetByKey(key)
	if err != nil {
		return errors.Logf(err, "try to get obj: %s return error", key)
	}

	ns, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		log.Error(err)
		return err
	}
	clusterName := strings.TrimSuffix(name, "-pdset")
	if !exists {
		log.Infof("TidbSet: [%s] has been deleted", key)
		_, err = pc.cli.PingcapV1().TidbSets(ns).Get(name, metav1.GetOptions{})
		if err != nil && apierrors.IsNotFound(err) {
			log.Infof("TidbSet: [%s/%s] has been deleted, deleting all related pd pods", ns, name)
			return pc.deletePDPods(ns, name)
		}

		return errors.Trace(err)
	}

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

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

	tc, err := pc.cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}

	tc, syncErr := pc.syncPDNamesFor(tc)
	if syncErr != nil {
		return errors.Logf(syncErr, "failed to sync pd names for TidbCluster: [%s/%s]", ns, clusterName)
	}

	if tc.Annotations == nil || len(tc.Annotations[v1.AnnPDPodNames]) == 0 {
		return errors.Errorf("TidbCluster: [%s/%s]'s annotations: %s is nil or empty", ns, clusterName, v1.AnnPDPodNames)
	}

	ts, pods, err := pc.syncPDReplicasFor(ts)
	if err != nil {
		return errors.Errorf("failed to sync replicas for TidbSet: [%s/%s], %v", ns, tsName, err)
	}

	if ts.Status.Replicas == 0 {
		for i := 0; i < int(ts.Spec.Replicas-ts.Status.Replicas); i++ {
			pod, createErr := pc.createPDInstanceFor(ts, true)
			if createErr != nil {
				return errors.Logf(createErr, "Pod in [%s/%s] failed to create", ns, clusterName)
			}

			log.Infof("Pod: [%s/%s] created successfully.", ns, pod.Name)
		}
	}

	if len(pods) > 0 {
		err = pc.syncPDStatusFor(ts)
		if err != nil {
			return errors.Errorf("failed to sync PD status for TidbSet: [%s/%s], %v", ns, tsName, err)
		}
	}

	return pc.syncPDPodFor(tc, ts, pods)
}

func (pc *Controller) syncPDPodFor(tc *v1.TidbCluster, ts *v1.TidbSet, pods []corev1.Pod) error {
	ns := ts.GetNamespace()
	clusterName := ts.OwnerReferences[0].Name

	for _, pod := range pods {
		podName := pod.GetName()
		if pod.Annotations[label.AnnShouldDeletedPD] == podName {
			msg := fmt.Sprintf("Pod: [%s/%s] was marked should deleted", ns, podName)
			log.Info(msg)

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

			return pc.deletePDPodFor(ts, &pod)
		}
	}

	annPodNames := strings.Split(tc.Annotations[v1.AnnPDPodNames], ",")

	shouldAdd := ts.Status.NormalReplicas == ts.Status.ReadyReplicas &&
		ts.Status.ReadyReplicas > ts.Status.Replicas/2 &&
		ts.Status.NormalReplicas < ts.Spec.Replicas
	if tc.ReallyRestoring() {
		shouldAdd = ts.Status.Replicas < ts.Spec.Replicas
	}
	if shouldAdd {
		pod, err := pc.createPDInstanceFor(ts, false)
		if err != nil {
			return errors.Logf(err, "Pod in [%s/%s] failed to create", ns, clusterName)
		}

		log.Infof("Pod[%s/%s] created successfully.", ns, pod.Name)
		return nil
	}

	if ts.Status.NormalReplicas > ts.Spec.Replicas {
		var pod *corev1.Pod
		var podNames []string

		sort.Sort(tsctl.ActivePods(pods))

	outLabel:
		for _, p := range pods {
			pn := p.GetName()
			podNames = append(podNames, pn)
			for _, annPodName := range annPodNames {
				if pn == annPodName {
					continue outLabel
				}
			}
			pod = &p
			break
		}

		if pod == nil {
			return errors.Errorf("TidbSet: [%s/%s], TidbCluster: [%s/%s] annPodNames: %v, podNames: %v, can't find a pod to deleting", ns, ts.GetName(), ns, clusterName, annPodNames, podNames)
		}

		err := review.Review(pc.cli, review.Info{
			UID:         string(pod.UID),
			Namespace:   ns,
			ClusterName: clusterName,
			Reason:      "DecreasePDReplicas",
			Message: fmt.Sprintf("TidbSet: [%s/%s]'s NormalReplicas: [%d] is greater than Replicas: [%d], deleting pod: [%s/%s]",
				ns, ts.GetName(), ts.Status.NormalReplicas, ts.Status.Replicas, ns, pod.GetName()),
		})
		if err != nil {
			return errors.Trace(err)
		}

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

		return pc.deletePDPodFor(ts, pod)
	}

	newImage := tsctl.GetImageByPodSpec(v1.PDContainerType, &ts.Spec.Template.Spec)
	for _, pod := range pods {
		podName := pod.GetName()
		oldImage := tsctl.GetImageByPodSpec(v1.PDContainerType, &pod.Spec)
		if oldImage != newImage && ts.Status.ReadyReplicas == ts.Status.Replicas {
			tsctl.SetImageByPodSpec(v1.PDContainerType, &pod.Spec, newImage)
			_, err := pc.kubeCli.CoreV1().Pods(ns).Update(&pod)
			if err != nil {
				return errors.Logf(err, "Pod: [%s/%s] upgrade to image: [%s] failed", ns, podName, newImage)
			}

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

	return nil
}

func (pc *Controller) createPVCForPDPod(tidbset *v1.TidbSet, podName string) (*corev1.PersistentVolumeClaim, error) {
	ns := tidbset.Namespace
	clusterName := tidbset.OwnerReferences[0].Name
	pvcList, listErr := pc.kubeCli.CoreV1().PersistentVolumeClaims(ns).List(label.ClusterListOptions(clusterName))
	if listErr != nil {
		log.Errorf("failed to list PVC in [%s/%s]: %v", ns, clusterName, listErr)
		return nil, errors.Trace(listErr)
	}
	for _, pvc := range pvcList.Items {
		if pvc.Annotations[label.AnnPodNameKey] == podName {
			log.Infof("PVC associated with pod[%s/%s] already exists", ns, podName)
			return &pvc, nil
		}
	}

	pvcName := fmt.Sprintf("pvc-%s-%s", tidbset.Name, uuid.New())
	storage := defaultPDRequestSize
	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, %v",
					storageFromAnnotation,
					pvcName,
					tidbset.Name,
					err,
				)
			}
			storage = q
		}
	}

	storageClassName := volumeprovisioner.StorageClassName
	newPVC := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      pvcName,
			Namespace: tidbset.Namespace,
			Labels:    label.New().Cluster(clusterName).PD(),
			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 := pc.kubeCli.CoreV1().PersistentVolumeClaims(tidbset.Namespace).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 (pc *Controller) getAPDName(tidbset *v1.TidbSet) (*string, []string, error) {
	ns := tidbset.GetNamespace()
	clusterName := tidbset.OwnerReferences[0].Name

	tc, err := pc.cli.PingcapV1().TidbClusters(ns).Get(clusterName, metav1.GetOptions{})
	if err != nil {
		return nil, nil, errors.Trace(err)
	}

	if tc.Annotations == nil || len(tc.Annotations[v1.AnnPDPodNames]) == 0 {
		return nil, nil, errors.Errorf("TidbCluster: [%s/%s]'s annotations: %s is nil or empty", ns, clusterName, v1.AnnPDPodNames)
	}
	annPodNames := strings.Split(tc.Annotations[v1.AnnPDPodNames], ",")

	var usedPodNames []string
	podListOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(tidbset.Spec.Selector.MatchLabels).String(),
	}
	podList, err := pc.kubeCli.CoreV1().Pods(ns).List(podListOptions)
	if err != nil {
		return nil, annPodNames, errors.Trace(err)
	}
	for _, pod := range podList.Items {
		usedPodNames = append(usedPodNames, pod.Name)
	}

	var podName string
nextNameLabel:
	for _, annPodName := range annPodNames {
		for _, usedPodName := range usedPodNames {
			if annPodName == usedPodName {
				continue nextNameLabel
			}
		}

		podName = annPodName
	}

	if len(podName) == 0 {
		return nil, annPodNames, errors.Errorf("can't generate a pod name, annPodNames: %v, usedPodNames: %v", annPodNames, usedPodNames)
	}

	return &podName, annPodNames, nil
}

func (pc *Controller) getPeers(tidbset *v1.TidbSet) []string {
	urls := []string{}
	clusterName := strings.TrimSuffix(tidbset.OwnerReferences[0].Name, "-pdset")
	ns := tidbset.Namespace
	opts := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(tidbset.Spec.Selector.MatchLabels).String()}
	podList, err := pc.kubeCli.CoreV1().Pods(ns).List(opts)
	if err != nil {
		log.Errorf("failed to list PD pod in [%s/%s]: %v", ns, clusterName, err)
		return urls
	}
	for _, pod := range podList.Items {
		// skip pod without hostname or subdomain, which is created by old controller-manager
		if pod.Spec.Hostname == "" || pod.Spec.Subdomain == "" {
			continue
		}
		if pod.Status.Phase == corev1.PodRunning {
			u := fmt.Sprintf("http://%s.%s-pd-peer.%s.svc:2379", pod.Name, clusterName, ns)
			urls = append(urls, u)
		}
	}
	return urls
}

func (pc *Controller) createPDInstanceFor(ts *v1.TidbSet, init bool) (*corev1.Pod, error) {
	ns := ts.Namespace
	clusterName := ts.OwnerReferences[0].Name
	podName, allPodNames, err := pc.getAPDName(ts)
	if err != nil {
		return nil, errors.Trace(err)
	}
	pvc, err := pc.createPVCForPDPod(ts, *podName)
	if err != nil {
		return nil, errors.Trace(err)
	}
	uid := ts.UID
	controller := true
	ownerRef := metav1.OwnerReference{
		APIVersion: tsctl.ControllerKind.GroupVersion().String(),
		Kind:       tsctl.ControllerKind.Kind,
		Name:       ts.Name,
		UID:        uid,
		Controller: &controller,
	}
	podSpec := ts.Spec.Template.Spec.DeepCopy()

	podSpec.Hostname = *podName
	args := podSpec.Containers[0].Args
	if init {
		initialPeers := make([]string, 0, len(allPodNames))
		for _, n := range allPodNames {
			initialPeers = append(initialPeers, fmt.Sprintf("%s=http://%s.%s-pd-peer.%s.svc:2380", n, n, clusterName, ns))
		}
		initialClusterURLs := fmt.Sprintf("--initial-cluster=%s", strings.Join(initialPeers, ","))
		podSpec.Containers[0].Args = append(args, initialClusterURLs)
	} else {
		peers := pc.getPeers(ts)
		if len(peers) == 0 {
			// if no pods have hostname and subdomain, we still use pd client service to join new pod
			peers = append(peers, fmt.Sprintf("http://%s-pd:2379", clusterName))
		}
		joinURLs := fmt.Sprintf("--join=%s", strings.Join(peers, ","))
		podSpec.Containers[0].Args = append(args, joinURLs)
	}
	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:          label.New().Cluster(clusterName).PD(),
			OwnerReferences: []metav1.OwnerReference{ownerRef},
			Annotations:     util.AnnProm(nil, "/metrics"),
		},
		Spec: *podSpec,
	}
	return pc.kubeCli.CoreV1().Pods(ns).Create(pod)
}

func (pc *Controller) deletePDPods(ns, name string) error {
	listOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(strings.TrimSuffix(name, "-pdset")).PD().Labels(),
		).String(),
	}
	podList, err := pc.kubeCli.CoreV1().Pods(ns).List(listOptions)
	if err != nil {
		return errors.Trace(err)
	}

	for _, pod := range podList.Items {
		podName := pod.GetName()
		err = pc.releasePVCFor(&pod)
		if err != nil {
			return errors.Logf(err, "failed to release PVC for Pod: [%s/%s]", ns, podName)
		}

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

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

	return nil
}

// deregister pd from pd cluster, caller get pdName from pods, it's always non-empty
func (pc *Controller) deletePDByName(namespace, clusterName, pdName string) error {
	pdCli := util.NewPDClientForCluster(namespace, clusterName, timeout)
	err := pdCli.DeleteMember(pdName)
	if err != nil {
		return errors.Logf(err, "PD member[%s/%s] from cluster[%s] failed to delete", namespace, pdName, clusterName)
	}
	log.Infof("PD member[%s/%s] from cluster[%s] deleted successfully.", namespace, pdName, clusterName)
	return nil
}

func (pc *Controller) deletePDByID(namespace, clusterName string, id string) error {
	pdCli := util.NewPDClientForCluster(namespace, clusterName, timeout)
	err := pdCli.DeleteMemberByID(id) // member info get from pd may have no name
	if err != nil {
		return errors.Logf(err, "PD member[%s/%s] from cluster[%s] failed to delete", namespace, id, clusterName)
	}
	log.Infof("PD member[%s/%d] from cluster[%s] deleted successfully.", namespace, id, clusterName)
	return nil
}

func (pc *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.AnnShouldDeletedPD] != podName {
		pod.Annotations[label.AnnShouldDeletedPD] = podName
		_, err := pc.kubeCli.CoreV1().Pods(ns).Update(pod)
		if err != nil {
			return errors.Logf(err, "failed to update Pod: [%s/%s]'s annotations: [%s] to: [%s]", ns, podName, label.AnnShouldDeletedPD, podName)
		}

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

	return nil
}

func (pc *Controller) deletePDPodFor(ts *v1.TidbSet, pod *corev1.Pod) error {
	ns := ts.GetNamespace()
	clusterName := ts.OwnerReferences[0].Name
	podName := pod.GetName()
	member, ok := ts.Status.PDStatus[podName]
	if ok {
		memberID := member.ID
		if len(memberID) == 0 {
			return errors.Errorf("TidbSet: [%s/%s]'s Pod: [%s/%s] memberID is empty", ns, ts.GetName(), ns, podName)
		}

		err := pc.deletePDByID(ns, clusterName, memberID)
		if err != nil {
			return errors.Trace(err)
		}
	}

	err := pc.markPVTombstoneAndReleasePVCFor(pod)
	if err != nil {
		return errors.Trace(err)
	}
	err = pc.kubeCli.CoreV1().Pods(ns).Delete(podName, nil)
	if err != nil {
		return errors.Logf(err, "failed to delete Pod: [%s/%s]", ns, podName)
	}
	log.Infof("Pod[%s/%s] deleted successfully.", ns, podName)
	return nil
}

func (pc *Controller) syncPDReplicasFor(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, "-pdset")

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

	nodes, err := pc.kubeCli.CoreV1().Nodes().List(metav1.ListOptions{})
	if err != nil {
		return ts, nil, errors.Trace(err)
	}

	for _, pod := range podList.Items {
		replicas++
		podName := pod.GetName()
		if tsctl.PodInUnreachableNode(&pod, nodes.Items) {
			msg := fmt.Sprintf("Pod: [%s/%s]'s node: [%s] is unreachable, deleting it from PD cluster",
				ns, podName, pod.Spec.NodeName)
			log.Info(msg)

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

			err = pc.deletePDByName(ns, clusterName, podName)
			if err != nil {
				return ts, podList.Items, errors.Trace(err)
			}

			continue
		}

		if pod.Annotations == nil {
			pod.Annotations = make(map[string]string)
		}
		ann := pod.Annotations[label.AnnShouldDeletedPD]

		if ann == podName {
			log.Infof("Pod: [%s/%s]'s annotation[%s] is: [%s], skipping", ns, podName, label.AnnShouldDeletedPD, ann)
			continue
		}

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

	if ts.Status.Replicas == replicas &&
		ts.Status.NormalReplicas == normalReplicas &&
		ts.Status.ReadyReplicas == readyReplicas {
		log.Debugf(
			"TidbSet: [%s/%s] Status.Replicas = [%d] Status.NormalReplicas = [%d] Status.ReadyReplicas = [%d]"+
				"no need to update",
			ns, name, replicas, normalReplicas, readyReplicas)
		return ts, podList.Items, nil
	}

	ts.Status.Replicas = replicas
	ts.Status.NormalReplicas = normalReplicas
	ts.Status.ReadyReplicas = readyReplicas

	ts, err = pc.cli.PingcapV1().TidbSets(ns).Update(ts)
	if err != nil {
		log.Errorf("TidbSet[%s/%s] status update failed: %v", ns, name, err)
		return ts, podList.Items, errors.Trace(err)
	}
	log.Infof("TidbSet[%s/%s] status updated successfully.", ns, name)

	return ts, podList.Items, nil
}

func (pc *Controller) syncPDStatusFor(ts *v1.TidbSet) error {
	ns := ts.GetNamespace()
	tsName := ts.GetName()
	clusterName := strings.TrimSuffix(ts.OwnerReferences[0].Name, "-pdset")

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

	pdCli := util.NewPDClientForCluster(ns, clusterName, timeout)
	membersInfo, memberErr := pdCli.GetMembers()
	if memberErr != nil {
		return errors.Logf(memberErr, "failed to list PD members in [%s/%s]", ns, clusterName)
	}
	if membersInfo == nil {
		return errors.Errorf("pdCli.GetMembers return nil in [%s/%s]", ns, clusterName)
	}

	pdStatus := make(map[string]v1.PDStatus)

	for _, member := range membersInfo.Members {
		id := member.GetMemberId()
		memberID := fmt.Sprintf("%d", id)
		// FIXME PeerUrls may be empty?
		peerURL := member.PeerUrls[0]
		url, err := url.Parse(peerURL)
		if err != nil {
			return errors.Logf(err, "failed to parse peerUrl: [%s] in [%s/%s]", peerURL, ns, clusterName)
		}

		ip := strings.Split(url.Host, ":")[0]
		name := member.GetName()
		if len(name) == 0 {
			name = strings.Split(ip, ".")[0]
			log.Warnf("PD member: [%d] don't have a name, get its name from peerUrls: [%s], got name: [%v] in [%s/%s]",
				id, member.PeerUrls, name, ns, clusterName)
		}

		// may be
		if len(name) == 0 {
			log.Warnf("PD member: [%d] don't have a name, and can't get it from peerUrls: [%s], memberInfo: [%v] in [%s/%s]",
				id, member.PeerUrls, member, ns, clusterName)
			continue
		}

		status := v1.PDStatus{
			Name: name,
			ID:   memberID,
			IP:   ip,
		}

		var pdPod *corev1.Pod
		for _, pod := range podList.Items {
			if pod.GetName() == name {
				pdPod = &pod
				break
			}
		}

		if pdPod == nil {
			continue
		}

		pdStatus[name] = status
	}

	if !reflect.DeepEqual(pdStatus, ts.Status.PDStatus) {
		ts.Status.PDStatus = pdStatus
		if _, err := pc.cli.PingcapV1().TidbSets(ns).Update(ts); err != nil {
			return errors.Logf(err, "failed to update TidbSet[%s/%s] status", ns, tsName)
		}

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

	return nil
}

func (pc *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 := pc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Get(claimName, metav1.GetOptions{})
			if err != nil {
				return errors.Trace(err)
			}

			pvName := pvc.Spec.VolumeName
			pv, err := pc.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 = pc.kubeCli.CoreV1().PersistentVolumes().Update(pv)
			if err != nil {
				return errors.Logf(err, "failed to set PV: %s annotation: %s", pvName, volumeprovisioner.AnnPVTombstoned)
			}
			log.Infof("Set PV: %s annotation: %s to %s successfully", pvName, volumeprovisioner.AnnPVTombstoned, now)

			err = pc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(claimName, nil)
			if err != nil && !apierrors.IsNotFound(err) {
				return errors.Logf(err, "Pod: [%s/%s]'s PVC: [%s/%s] released failed", ns, podName, ns, claimName)
			}
			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 (pc *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 := pc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(claimName, nil)
			if err != nil && !apierrors.IsNotFound(err) {
				return errors.Logf(err, "Pod: [%s/%s]'s PVC: [%s/%s] released failed", ns, podName, ns, claimName)
			}
			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 (pc *Controller) syncPDNamesFor(tc *v1.TidbCluster) (*v1.TidbCluster, error) {
	ns := tc.GetNamespace()
	clusterName := tc.GetName()

	if tc.Annotations == nil {
		tc.Annotations = make(map[string]string)
	}

	var annPodNames []string
	ann := tc.Annotations[v1.AnnPDPodNames]
	if len(ann) != 0 {
		annPodNames = strings.Split(ann, ",")
	}

	podListOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(label.New().Cluster(clusterName).PD().Labels()).String(),
	}
	podList, err := pc.kubeCli.CoreV1().Pods(ns).List(podListOptions)
	if err != nil {
		return tc, errors.Trace(err)
	}

	nodes, err := pc.kubeCli.CoreV1().Nodes().List(metav1.ListOptions{})
	if err != nil {
		return tc, errors.Trace(err)
	}

	var deleteAnnPodNames []string
	for _, pod := range podList.Items {
		if tsctl.PodInUnreachableNode(&pod, nodes.Items) {
			log.Warningf("TidbCluster: [%s/%s]'s pd pod: [%s] is evicted,now delete it's name from tidbcluster's annotations[%s]", ns, clusterName, pod.GetName(), v1.AnnPDPodNames)
			deleteAnnPodNames = append(deleteAnnPodNames, pod.GetName())
		}
	}
	if len(deleteAnnPodNames) > 0 {
		annPodNames = util.FilterSlice(annPodNames, deleteAnnPodNames)
	}

	if len(annPodNames) == int(tc.Spec.PD.Size) {
		log.Debugf("TidbCluster: [%s/%s]'s annotations: %s's size equal PD size: %d", ns, clusterName, v1.AnnPDPodNames, tc.Spec.PD.Size)
		return tc, nil
	}

	// forward compatible
	if len(annPodNames) == 0 && len(podList.Items) != 0 {
		for _, pod := range podList.Items {
			if !tsctl.PodInUnreachableNode(&pod, nodes.Items) {
				annPodNames = append(annPodNames, pod.GetName())
			}
		}
	}
	log.Infof("TidbCluster: [%s/%s]'s current annPodNames is %v", ns, clusterName, annPodNames)

	if len(annPodNames) > int(tc.Spec.PD.Size) {
		tc.Annotations[v1.AnnPDPodNames] = strings.Join(annPodNames[:tc.Spec.PD.Size], ",")
	} else {
		for i := 0; i < int(tc.Spec.PD.Size)-len(annPodNames); i++ {
			retryCount := 0
		retryLabel:
			if retryCount >= 3 {
				return tc, errors.Errorf("Three consecutive generation of the same pod name, TidbCluster: [%s/%s]", ns, clusterName)
			}

			retryCount++
			newName := namegenerator.SimpleNameGenerator.GenerateName(fmt.Sprintf("%s-pd-", clusterName))

			for _, n := range annPodNames {
				if n == newName {
					goto retryLabel
				}
			}
			annPodNames = append(annPodNames, newName)
			log.Infof("TidbCluster: [%s/%s] add a new pd name: %s", ns, clusterName, newName)
		}

		tc.Annotations[v1.AnnPDPodNames] = strings.Join(annPodNames, ",")
	}

	tc, err = pc.cli.PingcapV1().TidbClusters(ns).Update(tc)
	if err != nil {
		log.Errorf("failed to update TidbCluster: [%s/%s] when sync PD names, %v", ns, clusterName, err)
		return tc, errors.Trace(err)
	}

	log.Infof("update TidbCluster: [%s/%s] successfully when sync PD names", ns, clusterName)
	return tc, nil
}