func (pmm *pdMemberManager) syncPDConfigMap(tc *v1alpha1.TikvCluster, set *apps.StatefulSet) (*corev1.ConfigMap, error) {
	if tc.Spec.PD.Config == nil {
		return nil, nil
	}
	newCm, err := getPDConfigMap(tc)
	if err != nil {
		return nil, err
	}
	if set != nil && tc.BasePDSpec().ConfigUpdateStrategy() == v1alpha1.ConfigUpdateStrategyInPlace {
		inUseName := FindConfigMapVolume(&set.Spec.Template.Spec, func(name string) bool {
			return strings.HasPrefix(name, controller.PDMemberName(tc.Name))
		})
		if inUseName != "" {
			newCm.Name = inUseName
		}
	}
	return pmm.typedControl.CreateOrUpdateConfigMap(tc, newCm)
}
syncPDConfigMap syncs the configmap of PD
package member
import (
	"fmt"
	"strconv"
	"strings"
	"github.com/Masterminds/semver"
	"github.com/tikv/tikv-operator/pkg/apis/tikv/v1alpha1"
	"github.com/tikv/tikv-operator/pkg/controller"
	"github.com/tikv/tikv-operator/pkg/label"
	"github.com/tikv/tikv-operator/pkg/manager"
	"github.com/tikv/tikv-operator/pkg/pdapi"
	"github.com/tikv/tikv-operator/pkg/util"
	apps "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/apimachinery/pkg/util/uuid"
	v1 "k8s.io/client-go/listers/apps/v1"
	corelisters "k8s.io/client-go/listers/core/v1"
	"k8s.io/klog"
	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
)
const (
	pdClusterCertPath  = "/var/lib/pd-tls"
	tidbClientCertPath = "/var/lib/tidb-client-tls"
)
type pdMemberManager struct {
	pdControl    pdapi.PDControlInterface
	setControl   controller.StatefulSetControlInterface
	svcControl   controller.ServiceControlInterface
	podControl   controller.PodControlInterface
	typedControl controller.TypedControlInterface
	setLister    v1.StatefulSetLister
	svcLister    corelisters.ServiceLister
	podLister    corelisters.PodLister
	epsLister    corelisters.EndpointsLister
	pvcLister    corelisters.PersistentVolumeClaimLister
	pdScaler     Scaler
	pdUpgrader   Upgrader
	autoFailover bool
	pdFailover   Failover
}
func NewPDMemberManager(pdControl pdapi.PDControlInterface,
	setControl controller.StatefulSetControlInterface,
	svcControl controller.ServiceControlInterface,
	podControl controller.PodControlInterface,
	typedControl controller.TypedControlInterface,
	setLister v1.StatefulSetLister,
	svcLister corelisters.ServiceLister,
	podLister corelisters.PodLister,
	epsLister corelisters.EndpointsLister,
	pvcLister corelisters.PersistentVolumeClaimLister,
	pdScaler Scaler,
	pdUpgrader Upgrader,
	autoFailover bool,
	pdFailover Failover) manager.Manager {
	return &pdMemberManager{
		pdControl,
		setControl,
		svcControl,
		podControl,
		typedControl,
		setLister,
		svcLister,
		podLister,
		epsLister,
		pvcLister,
		pdScaler,
		pdUpgrader,
		autoFailover,
		pdFailover}
}
func (pmm *pdMemberManager) Sync(tc *v1alpha1.TikvCluster) error {
	if err := pmm.syncPDServiceForTikvCluster(tc); err != nil {
		return err
	}
	if err := pmm.syncPDHeadlessServiceForTikvCluster(tc); err != nil {
		return err
	}
	return pmm.syncPDStatefulSetForTikvCluster(tc)
}
func (pmm *pdMemberManager) syncPDServiceForTikvCluster(tc *v1alpha1.TikvCluster) error {
	if tc.Spec.Paused {
		klog.V(4).Infof("tidb cluster %s/%s is paused, skip syncing for pd service", tc.GetNamespace(), tc.GetName())
		return nil
	}
	ns := tc.GetNamespace()
	tcName := tc.GetName()
	newSvc := pmm.getNewPDServiceForTikvCluster(tc)
	oldSvcTmp, err := pmm.svcLister.Services(ns).Get(controller.PDMemberName(tcName))
	if errors.IsNotFound(err) {
		err = controller.SetServiceLastAppliedConfigAnnotation(newSvc)
		if err != nil {
			return err
		}
		return pmm.svcControl.CreateService(tc, newSvc)
	}
	if err != nil {
		return err
	}
	oldSvc := oldSvcTmp.DeepCopy()
	equal, err := controller.ServiceEqual(newSvc, oldSvc)
	if err != nil {
		return err
	}
	if !equal {
		svc := *oldSvc
		svc.Spec = newSvc.Spec
		err = controller.SetServiceLastAppliedConfigAnnotation(&svc)
		if err != nil {
			return err
		}
		svc.Spec.ClusterIP = oldSvc.Spec.ClusterIP
		_, err = pmm.svcControl.UpdateService(tc, &svc)
		return err
	}
	return nil
}
func (pmm *pdMemberManager) syncPDHeadlessServiceForTikvCluster(tc *v1alpha1.TikvCluster) error {
	if tc.Spec.Paused {
		klog.V(4).Infof("tidb cluster %s/%s is paused, skip syncing for pd headless service", tc.GetNamespace(), tc.GetName())
		return nil
	}
	ns := tc.GetNamespace()
	tcName := tc.GetName()
	newSvc := getNewPDHeadlessServiceForTikvCluster(tc)
	oldSvc, err := pmm.svcLister.Services(ns).Get(controller.PDPeerMemberName(tcName))
	if errors.IsNotFound(err) {
		err = controller.SetServiceLastAppliedConfigAnnotation(newSvc)
		if err != nil {
			return err
		}
		return pmm.svcControl.CreateService(tc, newSvc)
	}
	if err != nil {
		return err
	}
	equal, err := controller.ServiceEqual(newSvc, oldSvc)
	if err != nil {
		return err
	}
	if !equal {
		svc := *oldSvc
		svc.Spec = newSvc.Spec
		err = controller.SetServiceLastAppliedConfigAnnotation(&svc)
		if err != nil {
			return err
		}
		_, err = pmm.svcControl.UpdateService(tc, &svc)
		return err
	}
	return nil
}
func (pmm *pdMemberManager) syncPDStatefulSetForTikvCluster(tc *v1alpha1.TikvCluster) error {
	ns := tc.GetNamespace()
	tcName := tc.GetName()
	oldPDSetTmp, err := pmm.setLister.StatefulSets(ns).Get(controller.PDMemberName(tcName))
	if err != nil && !errors.IsNotFound(err) {
		return err
	}
	setNotExist := errors.IsNotFound(err)
	oldPDSet := oldPDSetTmp.DeepCopy()
	if err := pmm.syncTikvClusterStatus(tc, oldPDSet); err != nil {
		klog.Errorf("failed to sync TikvCluster: [%s/%s]'s status, error: %v", ns, tcName, err)
	}
	if tc.Spec.Paused {
		klog.V(4).Infof("tidb cluster %s/%s is paused, skip syncing for pd statefulset", tc.GetNamespace(), tc.GetName())
		return nil
	}
	cm, err := pmm.syncPDConfigMap(tc, oldPDSet)
	if err != nil {
		return err
	}
	newPDSet, err := getNewPDSetForTikvCluster(tc, cm)
	if err != nil {
		return err
	}
	if setNotExist {
		err = SetStatefulSetLastAppliedConfigAnnotation(newPDSet)
		if err != nil {
			return err
		}
		if err := pmm.setControl.CreateStatefulSet(tc, newPDSet); err != nil {
			return err
		}
		tc.Status.PD.StatefulSet = &apps.StatefulSetStatus{}
		return controller.RequeueErrorf("TikvCluster: [%s/%s], waiting for PD cluster running", ns, tcName)
	}
	if !tc.Status.PD.Synced {
		force := NeedForceUpgrade(tc)
		if force {
			tc.Status.PD.Phase = v1alpha1.UpgradePhase
			setUpgradePartition(newPDSet, 0)
			errSTS := updateStatefulSet(pmm.setControl, tc, newPDSet, oldPDSet)
			return controller.RequeueErrorf("tidbcluster: [%s/%s]'s pd needs force upgrade, %v", ns, tcName, errSTS)
		}
	}
	if !templateEqual(newPDSet, oldPDSet) || tc.Status.PD.Phase == v1alpha1.UpgradePhase {
		if err := pmm.pdUpgrader.Upgrade(tc, oldPDSet, newPDSet); err != nil {
			return err
		}
	}
	if err := pmm.pdScaler.Scale(tc, oldPDSet, newPDSet); err != nil {
		return err
	}
	if pmm.autoFailover {
		if pmm.shouldRecover(tc) {
			pmm.pdFailover.Recover(tc)
		} else if tc.PDAllPodsStarted() && !tc.PDAllMembersReady() || tc.PDAutoFailovering() {
			if err := pmm.pdFailover.Failover(tc); err != nil {
				return err
			}
		}
	}
	return updateStatefulSet(pmm.setControl, tc, newPDSet, oldPDSet)
}
func (pmm *pdMemberManager) shouldRecover(tc *v1alpha1.TikvCluster) bool {
	if tc.Status.PD.FailureMembers == nil {
		return false
	}
	for ordinal := range tc.PDStsDesiredOrdinals(true) {
		name := fmt.Sprintf("%s-%d", controller.PDMemberName(tc.GetName()), ordinal)
		pod, err := pmm.podLister.Pods(tc.Namespace).Get(name)
		if err != nil {
			klog.Errorf("pod %s/%s does not exist: %v", tc.Namespace, name, err)
			return false
		}
		if !podutil.IsPodReady(pod) {
			return false
		}
		status, ok := tc.Status.PD.Members[pod.Name]
		if !ok || !status.Health {
			return false
		}
	}
	return true
}
func (pmm *pdMemberManager) syncTikvClusterStatus(tc *v1alpha1.TikvCluster, set *apps.StatefulSet) error {
	if set == nil {
		return nil
	}
	ns := tc.GetNamespace()
	tcName := tc.GetName()
	tc.Status.PD.StatefulSet = &set.Status
	upgrading, err := pmm.pdStatefulSetIsUpgrading(set, tc)
	if err != nil {
		return err
	}
	if upgrading {
		tc.Status.PD.Phase = v1alpha1.UpgradePhase
	} else {
		tc.Status.PD.Phase = v1alpha1.NormalPhase
	}
	pdClient := controller.GetPDClient(pmm.pdControl, tc)
	healthInfo, err := pdClient.GetHealth()
	if err != nil {
		tc.Status.PD.Synced = false
		eps, epErr := pmm.epsLister.Endpoints(ns).Get(controller.PDMemberName(tcName))
		if epErr != nil {
			return fmt.Errorf("%s, %s", err, epErr)
		}
		if eps != nil && len(eps.Subsets) == 0 {
			return fmt.Errorf("%s, service %s/%s has no endpoints", err, ns, controller.PDMemberName(tcName))
		}
		return err
	}
	cluster, err := pdClient.GetCluster()
	if err != nil {
		tc.Status.PD.Synced = false
		return err
	}
	tc.Status.ClusterID = strconv.FormatUint(cluster.Id, 10)
	leader, err := pdClient.GetPDLeader()
	if err != nil {
		tc.Status.PD.Synced = false
		return err
	}
	pdStatus := map[string]v1alpha1.PDMember{}
	for _, memberHealth := range healthInfo.Healths {
		id := memberHealth.MemberID
		memberID := fmt.Sprintf("%d", id)
		var clientURL string
		if len(memberHealth.ClientUrls) > 0 {
			clientURL = memberHealth.ClientUrls[0]
		}
		name := memberHealth.Name
		if len(name) == 0 {
			klog.Warningf("PD member: [%d] doesn't have a name, and can't get it from clientUrls: [%s], memberHealth Info: [%v] in [%s/%s]",
				id, memberHealth.ClientUrls, memberHealth, ns, tcName)
			continue
		}
		status := v1alpha1.PDMember{
			Name:      name,
			ID:        memberID,
			ClientURL: clientURL,
			Health:    memberHealth.Health,
		}
		oldPDMember, exist := tc.Status.PD.Members[name]
		status.LastTransitionTime = metav1.Now()
		if exist && status.Health == oldPDMember.Health {
			status.LastTransitionTime = oldPDMember.LastTransitionTime
		}
		pdStatus[name] = status
	}
	tc.Status.PD.Synced = true
	tc.Status.PD.Members = pdStatus
	tc.Status.PD.Leader = tc.Status.PD.Members[leader.GetName()]
	tc.Status.PD.Image = ""
	c := filterContainer(set, "pd")
	if c != nil {
		tc.Status.PD.Image = c.Image
	}
	err = pmm.collectUnjoinedMembers(tc, set, pdStatus)
	if err != nil {
		return err
	}
	return nil
}