package cluster

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	reconcilerr "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/controllers/mixeddeploypd"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-api/pkg/provider"
	apiutil "github.com/tidbcloud/infra-api/pkg/util"
)

const (
	observeAfter = 1 * time.Minute

	// TiDB-Operator may miss to update `lastTransitionTime` of some components, so we add a skew to pods' creation time.
	podCreationTimeSkew = 5 * time.Minute
)

type TiDBClusterStatusResolver struct {
	// Provider specific
	client.Client
	provider.Machine
	provider.Cluster

	// Provider agnostic
	Target      *cloudv1.Cluster
	TiDBCluster *v1alpha1.TidbCluster

	Log logr.Logger

	PDMixedDeployConfig *cloudv1.MixedDeployPDWithTiKV
}

// Resolve ... TODO(aylei): we can make the observation more responsive via registering a controller to shoot api-server
func (r *TiDBClusterStatusResolver) Resolve(ctx context.Context) error {
	var errs error
	tidb, err := r.computeTiDBStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	pd, err := r.computePDStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tikv, err := r.computeTiKVStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tiflash, err := r.computeTiFlashStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tiproxy, err := r.computeTiProxyStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	if errs != nil {
		return errs
	}

	// Do not overwrite not own field.
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiDB != nil {
		tidb.AuditReady = r.Target.Status.DB.TiDB.AuditReady
	}

	var phase cloudv1.ClusterPhase
	secondaryStatus := cloudv1.DBSecondaryStatusHealthy

	// PD partial failure cannot be felt by user, so the status is hidden
	if tiproxy.Status == cloudv1.ComponentUnhealthy ||
		tikv.Status == cloudv1.ComponentUnhealthy ||
		tidb.Status == cloudv1.ComponentUnhealthy ||
		tiflash.Status == cloudv1.ComponentUnhealthy {
		secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
	}

	switch {
	// cluster is considered as unavailable when one of TiKV, PD, TiDB, TiProxy is unavailable
	case tidb.Status == cloudv1.ComponentUnavailable || pd.Status == cloudv1.ComponentUnavailable || tikv.Status == cloudv1.ComponentUnavailable || tiproxy.Status == cloudv1.ComponentUnavailable:
		if r.Target.Status.Phase != cloudv1.ClusterPhaseSuspend {
			phase = cloudv1.ClusterPhaseUnavailable
			secondaryStatus = cloudv1.DBSecondaryStatusUnhealthy
		} else {
			phase = cloudv1.ClusterPhaseSuspend
		}
	// if both upgrading and scaling are ongoing (e.g. failover when upgrading), show Upgrading
	case tidb.Status == cloudv1.ComponentUpgrading || tikv.Status == cloudv1.ComponentUpgrading || tiflash.Status == cloudv1.ComponentUpgrading || tiproxy.Status == cloudv1.ComponentUpgrading:
		phase = cloudv1.ClusterPhaseUpgrading
	case tidb.Status == cloudv1.ComponentScaling || tikv.Status == cloudv1.ComponentScaling || tiflash.Status == cloudv1.ComponentScaling || tiproxy.Status == cloudv1.ComponentScaling:
		phase = cloudv1.ClusterPhaseScaling
	default:
		phase = cloudv1.ClusterPhaseRunning
	}
	reason := fmt.Sprintf("pd: %s, tidb: %s, tikv: %s, tiflash: %s, tiproxy: %s", pd.Status, tidb.Status, tikv.Status, tiflash.Status, tiproxy.Status)
	dbStatus := &cloudv1.DBStatus{
		Version:         r.TiDBCluster.Spec.Version,
		TiDB:            tidb,
		TiKV:            tikv,
		PD:              pd,
		SecondaryStatus: secondaryStatus,
	}

	if r.Target.Spec.Cluster.TiFlash.Replicas > 0 {
		dbStatus.TiFlash = tiflash
	}
	if r.Target.Spec.Cluster.TiProxy.Replicas > 0 {
		dbStatus.TiProxy = tiproxy
	}

	if r.Target.Status.Conditions == nil {
		r.Target.Status.Conditions = []cloudv1.ClusterCondition{}
	}
	cloudv1.SetStatusCondition(&r.Target.Status.Conditions, r.buildSuspendCondition(dbStatus))

	if r.Target.Suspending() || r.Target.Suspended() {
		phase = cloudv1.ClusterPhaseSuspend
	}

	r.Target.Status.Phase = phase
	r.Target.Status.Reason = reason
	r.Target.Status.DB = dbStatus

	return nil
}
func (r *TiDBClusterStatusResolver) buildSuspendCondition(dbStatus *cloudv1.DBStatus) cloudv1.ClusterCondition {
	suspendCondition := cloudv1.ClusterCondition{
		Type:    cloudv1.ClusterSuspended,
		Status:  cloudv1.ConditionTrue,
		Reason:  "ClusterSuspended",
		Message: "The cluster has been suspended",
	}
	var msg []string
	if dbStatus == nil {
		msg = append(msg, "Cluster.Status.DB is nil")
	} else {
		if dbStatus.PD == nil {
			msg = append(msg, "Cluster.Status.DB.PD is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.PD.Conditions,
			cloudv1.ComponentSuspended) {
			msg = append(msg, "PD has not been suspended")
		}
		if dbStatus.TiDB == nil {
			msg = append(msg, "Cluster.Status.DB.TiDB is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.TiDB.Conditions,
			cloudv1.ComponentSuspended) {
			msg = append(msg, "TiDB has not been suspended")
		}
		if dbStatus.TiKV == nil {
			msg = append(msg, "Cluster.Status.DB.TiKV is nil")
		} else if !meta.IsStatusConditionTrue(dbStatus.TiKV.Conditions,
			cloudv1.ComponentSuspended) {
			msg = append(msg, "TiKV has not been suspended")
		}
		if r.Target.Spec.Cluster.TiFlash.Replicas > 0 {
			if dbStatus.TiFlash == nil {
				msg = append(msg, "Cluster.Status.DB.TiFlash is nil")
			} else if !meta.IsStatusConditionTrue(dbStatus.TiFlash.Conditions,
				cloudv1.ComponentSuspended) {
				msg = append(msg, "TiFlash has not been suspended")
			}
		}
	}
	if len(msg) > 0 {
		suspendCondition = cloudv1.ClusterCondition{
			Type:    cloudv1.ClusterSuspended,
			Status:  cloudv1.ConditionFalse,
			Reason:  "ClusterNotSuspended",
			Message: strings.Join(msg, ";"),
		}
	}
	return suspendCondition
}
func (r *TiDBClusterStatusResolver) computeComponentConditions(ctx context.Context, cType cloudv1.ComponentType, nodes []corev1.Node) []metav1.Condition {
	var conditions []metav1.Condition
	var suspendCondition metav1.Condition
	var needSuspend bool
	var component cloudv1.ClusterComponent
	memberType := utils.GetMemberType(cType)

	switch cType {
	case cloudv1.ComponentTiDB:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiDB != nil {
			conditions = r.Target.Status.DB.TiDB.Conditions
		}
		component = &r.Target.Spec.Cluster.TiDB

	case cloudv1.ComponentTiKV:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiKV != nil {
			conditions = r.Target.Status.DB.TiKV.Conditions
		}
		component = &r.Target.Spec.Cluster.TiKV
	case cloudv1.ComponentPD:
		if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
			conditions = r.Target.Status.DB.PD.Conditions
		}
		component = &r.Target.Spec.Cluster.PD
	case cloudv1.ComponentTiFlash:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiFlash != nil {
			conditions = r.Target.Status.DB.TiFlash.Conditions
		}
		component = &r.Target.Spec.Cluster.TiFlash
	case cloudv1.ComponentTiProxy:
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiProxy != nil {
			conditions = r.Target.Status.DB.TiProxy.Conditions
		}
		component = &r.Target.Spec.Cluster.TiProxy
	default:
		return nil
	}

	if conditions == nil {
		conditions = []metav1.Condition{}
	}
	needSuspend = component.NeedSuspend(r.Target.Spec.Cluster.SuspendAction)
	// The flow for the condition update is as below:
	// - If `suspendAction` is set in the spec, add the `ComponentSuspended`
	//   condition first to indicate that we're going to suspend the cluster
	// - Update TC with the `suspendAction` in the next reconcile,
	//   in case that the user may resume the cluster immediately
	//   after pausing the cluster
	// - Update `ComponentSuspended` condition to `True` after the cluster is suspended
	// - Remove the `ComponentSuspended` condition when the user resumes the cluster
	memberSuspending := r.TiDBCluster.ComponentIsSuspending(memberType)
	if !needSuspend && !memberSuspending {
		meta.RemoveStatusCondition(&conditions, cloudv1.ComponentSuspended)
		return conditions
	}
	memberSuspended := r.TiDBCluster.ComponentIsSuspended(memberType)
	nodeCount := len(nodes)

	if memberSuspended && nodeCount == 0 {
		suspendCondition = metav1.Condition{
			Type:    cloudv1.ComponentSuspended,
			Status:  metav1.ConditionTrue,
			Reason:  "ResourcesSuspended",
			Message: "All Pods and nodes are suspended",
		}
	} else {
		var msg []string
		if !memberSuspended {
			msg = append(msg, "Pods not suspended")
		}
		if nodeCount != 0 {
			msg = append(msg, fmt.Sprintf("%d Nodes not suspended", nodeCount))
		}
		suspendCondition = metav1.Condition{
			Type:    cloudv1.ComponentSuspended,
			Status:  metav1.ConditionFalse,
			Reason:  "ResourcesNotSuspended",
			Message: strings.Join(msg, ";"),
		}
	}
	meta.SetStatusCondition(&conditions, suspendCondition)
	return conditions
}

// selectComponentNodes selects nodes for the given component type
// For AWS self-managed nodegroup, after pausing the cluster, the nodes will still be schedulable for a few minutes.
// Refer to https://docs.aws.amazon.com/autoscaling/ec2/userguide/lifecycle-hooks-overview.html, the `Terminating:Wait` state will last for a few minutes.
// So we need to make sure that the nodes are deleted before we update the `ComponentSuspended` condition to `True`.
func (r *TiDBClusterStatusResolver) selectComponentNodes(ctx context.Context, componentType cloudv1.ComponentType, nodesOfPod []corev1.Node) ([]corev1.Node, error) {
	// If the cluster is not AWS self-managed nodegroup, use nodes of pods directly.
	if k8s.GetAnnotation(r.Target, consts.UseSelfManagedNodegroupAnnotation) != "true" {
		return nodesOfPod, nil
	}

	nodes, err := k8s.SelectNodes(ctx, r.Client, r.GetComponentNodeLabels(componentType))
	if err != nil {
		return nil, err
	}

	nodesMap := make(map[string]struct{})
	for _, node := range nodes {
		nodesMap[node.Name] = struct{}{}
	}

	for _, node := range nodesOfPod {
		if _, ok := nodesMap[node.Name]; !ok {
			nodes = append(nodes, node)
		}
	}

	return nodes, nil
}

func (r *TiDBClusterStatusResolver) computePDStatus(ctx context.Context) (*cloudv1.PDStatus, error) {
	var members []cloudv1.PDMember
	pdSpec := r.Target.Spec.Cluster.PD
	instanceSpec, err := r.FindInstanceType(pdSpec.GetInstanceType())
	if err != nil {
		return nil, err
	}

	pods, err := k8s.SelectPods(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(cloudv1.ComponentPD))
	if err != nil {
		return nil, err
	}
	nameToPod := make(map[string]corev1.Pod)
	for _, pod := range pods {
		nameToPod[pod.Name] = pod
	}

	// use pod labels as PVC labels now
	pvcs, err := k8s.SelectPVCs(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(cloudv1.ComponentPD))
	if err != nil {
		return nil, err
	}
	nameToPVC := make(map[string]corev1.PersistentVolumeClaim)
	for _, pvc := range pvcs {
		nameToPVC[pvcNameToPodName(pvc.Name)] = pvc
	}

	nodesOfPod := make([]corev1.Node, 0)
	for _, p := range pods {
		if len(p.Spec.NodeName) < 1 {
			continue
		}
		node := corev1.Node{}
		err := r.Client.Get(ctx, client.ObjectKey{
			Name: p.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, err
		}
		nodesOfPod = append(nodesOfPod, node)
	}

	nodes, err := r.selectComponentNodes(ctx, cloudv1.ComponentPD, nodesOfPod)
	if err != nil {
		return nil, err
	}
	conditions := r.computeComponentConditions(ctx, cloudv1.ComponentPD, nodes)

	var previousInstanceTypeMap map[string]string
	if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
		previousInstanceTypeMap = pdMemberToInstanceType(r.Target.Status.DB.PD.Members)
	}

	for name, status := range r.TiDBCluster.Status.PD.Members {
		memberStatus := cloudv1.MemberStatusNormal
		var healthAfterPodCreated bool
		if !status.Health {
			memberStatus = cloudv1.MemberStatusUnavailable
		} else if pod, ok := nameToPod[name]; ok {
			healthAfterPodCreated = status.LastTransitionTime.After(pod.CreationTimestamp.Time) ||
				time.Now().After(pod.CreationTimestamp.Add(podCreationTimeSkew))
		}

		var instanceType = previousInstanceTypeMap[name]
		if instanceType == "" || healthAfterPodCreated {
			// ref: https://github.com/pingcap/tidb-operator/blob/v1.3.6/pkg/controller/pod_control.go#L138
			podName := strings.Split(status.Name, ".")[0]
			if pod, ok := nameToPod[podName]; ok && pod.Spec.NodeName != "" {
				for _, node := range nodesOfPod {
					if node.Name == pod.Spec.NodeName {
						instanceType = r.GetNodeInstanceTypeFromLabels(node.Labels)
						break
					}
				}
			}
		}

		// use previous instance-type's CPU/Memory when scaling up to a new instance-type before ready
		if instanceType != "" {
			instanceSpec, err = r.FindInstanceType(instanceType)
			if err != nil {
				return nil, err
			}
		}

		// use the real PVC size when scaling
		storageSize := &pdSpec.StorageSize
		if pvc, ok := nameToPVC[name]; ok {
			pvcStorage := pvc.Spec.Resources.Requests[corev1.ResourceStorage]
			storageSize = &pvcStorage
		}

		members = append(members, cloudv1.PDMember{
			Name: name,
			Resources: cloudv1.Resources{
				CPU:     instanceSpec.CPU,
				Memory:  instanceSpec.Memory,
				Storage: storageSize,
			},
			Status:       memberStatus,
			InstanceType: instanceType,
		})
	}

	topo := r.Target.Spec.Cluster
	zeroPDNodes, mixedDeployPDWithTiKV, err := mixeddeploypd.ShouldUseZeroPDNodes(topo.PD.Replicas, topo.TiKV.Replicas, topo.TiDB.Replicas,
		topo.TiFlash.Replicas, "checking pd node scaling", r.TiDBCluster, r.Machine.(*cloudv1.Provider), r.GetComponentNodeLabels, r.Client, r.Target, r.PDMixedDeployConfig)
	if err != nil {
		return nil, err
	}
	pdNodeScaling := false
	if !mixedDeployPDWithTiKV {
		pdNodeScaling, err = r.componentNodeScaling(ctx, cloudv1.ComponentPD)
		if err != nil {
			return nil, err
		}
	} else {
		pdNodeScaling = !zeroPDNodes
	}

	volume, volModifying, err := r.componentVolumeStatus(ctx, &r.Target.Spec.Cluster.PD)
	if err != nil {
		return nil, err
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.PD != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.PD.InstanceTypeScalePhase
	}

	itScaling := false
	if !mixedDeployPDWithTiKV {
		// Instance of PD is not right when PD and TiKV are mixed deployed in one node, so skip checking instance type.
		itScaling = instanceTypeScalingFromPDMembers(members, pdSpec.GetInstanceType())
	}
	itScaling = itScaling || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	var status cloudv1.ComponentStatus
	switch {
	case !r.TiDBCluster.PDIsAvailable():
		status = cloudv1.ComponentUnavailable
	case r.TiDBCluster.PDUpgrading() && !itScaling: // we upgrading TidbCluster when scaling for instance type
		status = cloudv1.ComponentUpgrading
	case r.TiDBCluster.PDScaling() || pdNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !r.TiDBCluster.PDAllMembersReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if r.TiDBCluster.Status.PD.StatefulSet != nil {
		updatedReplicas = r.TiDBCluster.Status.PD.StatefulSet.UpdatedReplicas
	}

	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	return &cloudv1.PDStatus{
		Members:                members,
		Status:                 status,
		Volume:                 volume,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase, // do not overwrite it as it update by the cluster controller
		Conditions:             conditions,
	}, nil
}

func (r *TiDBClusterStatusResolver) computeTiDBStatus(ctx context.Context) (*cloudv1.TiDBStatus, error) {
	// Continue to refresh status if TiDB is suspended
	suspend := r.TiDBCluster.ComponentIsSuspended(v1alpha1.TiDBMemberType)
	// Skip refresh status if status in tc is outdated
	// like STS recreate.
	if !suspend &&
		len(r.TiDBCluster.Status.TiDB.Members) == 0 &&
		r.Target.Status.DB != nil &&
		r.Target.Status.DB.TiDB != nil {
		return r.Target.Status.DB.TiDB, nil
	}

	var members []cloudv1.TiDBMember
	tidbSpec := r.Target.Spec.Cluster.TiDB
	instanceSpec, err := r.FindInstanceType(tidbSpec.GetInstanceType())
	if err != nil {
		return nil, err
	}
	tidbNodeScaling, err := r.componentNodeScaling(ctx, cloudv1.ComponentTiDB)
	if err != nil {
		return nil, err
	}
	healthyCount := 0

	pods, err := k8s.SelectPods(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(cloudv1.ComponentTiDB))
	if err != nil {
		return nil, err
	}

	nodesOfPod := make([]corev1.Node, 0)
	for _, p := range pods {
		if len(p.Spec.NodeName) < 1 {
			continue
		}
		node := corev1.Node{}
		err := r.Client.Get(ctx, client.ObjectKey{
			Name: p.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, err
		}
		nodesOfPod = append(nodesOfPod, node)
	}

	// For suspend condition update, consider both
	// the TiDB nodesOfPod and the extra TiDB nodesOfPod
	nodesExtra, err := k8s.SelectNodes(ctx, r.Client, r.GetComponentNodeLabels(cloudv1.ComponentAdminTiDB))
	if err != nil {
		return nil, err
	}
	nodesExtra = append(nodesExtra, nodesOfPod...)

	nodes, err := r.selectComponentNodes(ctx, cloudv1.ComponentTiDB, nodesExtra)
	if err != nil {
		return nil, err
	}
	conditions := r.computeComponentConditions(ctx, cloudv1.ComponentTiDB, nodes)

	var nameToPod = make(map[string]corev1.Pod, len(pods))
	for _, pod := range pods {
		nameToPod[pod.Name] = pod
	}

	var previousInstanceTypeMap map[string]string
	var previousMemberStatusMap map[string]cloudv1.MemberStatus
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiDB != nil {
		previousInstanceTypeMap = tidbMemberToInstanceType(r.Target.Status.DB.TiDB.Members)
		previousMemberStatusMap = tidbMemberToMemberStatus(r.Target.Status.DB.TiDB.Members)
	}

	for name, status := range r.TiDBCluster.Status.TiDB.Members {
		pod, podOk := nameToPod[name]
		if !podOk {
			continue
		}

		var healthAfterPodCreated bool
		memberStatus := cloudv1.MemberStatusUnavailable
		switch {
		case pod.DeletionTimestamp != nil:
			memberStatus = cloudv1.MemberStatusTerminating
		case !k8s.IsPodReady(pod):
			memberStatus = cloudv1.MemberStatusCreating
		case !status.Health && previousMemberStatusMap[name] == cloudv1.MemberStatusCreating:
			memberStatus = cloudv1.MemberStatusCreating
		case status.Health && k8s.IsPodReady(pod):
			memberStatus = cloudv1.MemberStatusNormal

			healthyCount++
			healthAfterPodCreated = status.LastTransitionTime.After(pod.CreationTimestamp.Time) ||
				time.Now().After(pod.CreationTimestamp.Add(podCreationTimeSkew))
		}

		var zone string
		var instanceType = previousInstanceTypeMap[name]
		for _, node := range nodesOfPod {
			if node.Name == status.NodeName {
				zone = r.GetNodeZoneFromLabels(node.Labels)
				if instanceType == "" || healthAfterPodCreated {
					instanceType = r.GetNodeInstanceTypeFromLabels(node.Labels)
				}
				break
			}
		}

		if instanceType != "" {
			instanceSpec, err = r.FindInstanceType(instanceType)
			if err != nil {
				return nil, err
			}
		}

		members = append(members, cloudv1.TiDBMember{
			Name: name,
			Labels: map[string]string{
				cloudv1.MemberLabelAvailableZone: zone,
			},
			Resources: cloudv1.Resources{
				CPU:    instanceSpec.CPU,
				Memory: instanceSpec.Memory,
			},
			Status:       memberStatus,
			InstanceType: instanceType,
		})
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiDB != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiDB.InstanceTypeScalePhase
	}

	itScaling := instanceTypeScalingFromTiDBMembers(members, tidbSpec.GetInstanceType()) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	var status cloudv1.ComponentStatus
	switch {
	case healthyCount == 0:
		status = cloudv1.ComponentUnavailable
	case r.TiDBCluster.TiDBUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	case r.TiDBCluster.TiDBScaling() || tidbNodeScaling || itScaling:
		status = cloudv1.ComponentScaling
	case !r.TiDBCluster.TiDBAllMembersReady() || len(r.TiDBCluster.Status.TiDB.Members) != healthyCount:
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if r.TiDBCluster.Status.TiDB.StatefulSet != nil {
		updatedReplicas = r.TiDBCluster.Status.TiDB.StatefulSet.UpdatedReplicas
	}

	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	return &cloudv1.TiDBStatus{
		Members:                members,
		Status:                 status,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             conditions,
		TokenBasedAuthEnabled:  pointer.BoolDeref(r.TiDBCluster.Spec.TiDB.TokenBasedAuthEnabled, false),
	}, nil
}

func (r *TiDBClusterStatusResolver) computeTiProxyStatus(ctx context.Context) (*cloudv1.TiProxyStatus, error) {
	// if tiproxy is disabled on tidbcluster, report normal
	if !apiutil.IsTiProxyEnabled(r.TiDBCluster) {
		return &cloudv1.TiProxyStatus{Status: cloudv1.ComponentNormal}, nil
	}

	// compute scaling
	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiProxy != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiProxy.InstanceTypeScalePhase
	}
	itScaling := instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	// compute members and conditions
	var members []cloudv1.TiProxyMember
	var nodesOfPod []corev1.Node
	if r.TiDBCluster.TiProxyStsActualReplicas() != 0 {
		// get pods
		pods, err := k8s.SelectPods(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(cloudv1.ComponentTiProxy))
		if err != nil {
			return nil, err
		}
		var podMap = make(map[string]corev1.Pod, len(pods))
		for _, pod := range pods {
			podMap[pod.Name] = pod
		}

		// get nodesOfPod
		for _, p := range pods {
			if len(p.Spec.NodeName) < 1 {
				continue
			}
			node := corev1.Node{}
			err := r.Client.Get(ctx, client.ObjectKey{
				Name: p.Spec.NodeName,
			}, &node)
			if err != nil {
				return nil, err
			}
			nodesOfPod = append(nodesOfPod, node)
		}

		// previous status
		var previousInstanceTypeMap map[string]string
		if r.Target.Status.DB != nil && r.Target.Status.DB.TiProxy != nil {
			previousInstanceTypeMap = proxyMemberToInstanceType(r.Target.Status.DB.TiProxy.Members)
		}

		// compute member status
		for name, status := range r.TiDBCluster.Status.TiProxy.Members {
			member := cloudv1.TiProxyMember{
				Name:   name,
				Status: cloudv1.MemberStatusUnavailable,
			}

			pod, podOk := podMap[name]
			if podOk {
				switch {
				case pod.DeletionTimestamp != nil:
					member.Status = cloudv1.MemberStatusTerminating
				case pod.Status.Phase == corev1.PodPending:
					member.Status = cloudv1.MemberStatusCreating
				case status.Health:
					member.Status = cloudv1.MemberStatusNormal
				}

				var podInstanceType string
				if pod.Spec.NodeName != "" {
					for _, node := range nodesOfPod {
						if node.Name == pod.Spec.NodeName {
							podInstanceType = r.GetNodeInstanceTypeFromLabels(node.Labels)
							break
						}
					}
				}

				var instanceType string
				if instanceType == "" {
					// if there is no instanceType, try use the one from pod
					instanceType = podInstanceType
				} else if status.LastTransitionTime.After(pod.CreationTimestamp.Time) ||
					time.Now().After(pod.CreationTimestamp.Add(podCreationTimeSkew)) {
					// if it is up after the current pod, use the pod type
					instanceType = podInstanceType
				} else if instanceType != podInstanceType {
					// use previous instance-type's CPU/Memory when scaling up to a new instanc-type before ready
					instanceType = previousInstanceTypeMap[name]
				}

				if instanceType != "" {
					instanceSpec, err := r.FindInstanceType(instanceType)
					if err != nil {
						return nil, err
					}
					member.Resources.CPU = instanceSpec.CPU
					member.Resources.Memory = instanceSpec.Memory
					member.InstanceType = instanceType
				}
			}

			members = append(members, member)
		}
	}

	scaling, err := r.componentNodeScaling(ctx, cloudv1.ComponentTiProxy)
	if err != nil {
		return nil, err
	}

	var status cloudv1.ComponentStatus

	count := 0
	for _, member := range members {
		if member.Status == cloudv1.MemberStatusNormal {
			count++
		}
	}
	switch {
	case count == 0:
		status = cloudv1.ComponentUnavailable
	case r.TiDBCluster.Status.TiProxy.Phase == v1alpha1.UpgradePhase && !itScaling:
		status = cloudv1.ComponentUpgrading
	case r.TiDBCluster.Status.TiProxy.Phase == v1alpha1.ScalePhase || scaling || itScaling:
		status = cloudv1.ComponentScaling
	case count != len(members):
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if r.TiDBCluster.Status.TiProxy.StatefulSet != nil {
		updatedReplicas = r.TiDBCluster.Status.TiProxy.StatefulSet.UpdatedReplicas
	}

	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	nodes, err := r.selectComponentNodes(ctx, cloudv1.ComponentTiProxy, nodesOfPod)
	if err != nil {
		return nil, err
	}
	conditions := r.computeComponentConditions(ctx, cloudv1.ComponentTiProxy, nodes)

	// compute conditions
	return &cloudv1.TiProxyStatus{
		Members:                members,
		Status:                 status,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             conditions,
	}, nil
}

func (r *TiDBClusterStatusResolver) computeTiKVStatus(ctx context.Context) (*cloudv1.TiKVStatus, error) {
	tikvMembers, conditions, err := r.computeStoreMembers(ctx, &r.Target.Spec.Cluster.TiKV)
	if err != nil {
		return nil, err
	}
	tikvNodeScaling, err := r.componentNodeScaling(ctx, cloudv1.ComponentTiKV)
	if err != nil {
		return nil, err
	}
	volume, volModifying, err := r.componentVolumeStatus(ctx, &r.Target.Spec.Cluster.TiKV)
	if err != nil {
		return nil, err
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiKV != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiKV.InstanceTypeScalePhase
	}

	itScaling := instanceTypeScalingFromStoreMembers(tikvMembers, r.Target.Spec.Cluster.TiKV.InstanceType) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	var status cloudv1.ComponentStatus
	switch {
	case !r.TiDBCluster.TiKVIsAvailable():
		status = cloudv1.ComponentUnavailable
	case r.TiDBCluster.TiKVUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	case r.TiDBCluster.TiKVScaling() || tikvNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !r.TiDBCluster.TiKVAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if r.TiDBCluster.Status.TiKV.StatefulSet != nil {
		updatedReplicas = r.TiDBCluster.Status.TiKV.StatefulSet.UpdatedReplicas
	}

	sort.Slice(tikvMembers, func(i, j int) bool {
		return tikvMembers[i].Name < tikvMembers[j].Name
	})

	return &cloudv1.TiKVStatus{
		Members:                tikvMembers,
		Status:                 status,
		Volume:                 volume,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             conditions,
	}, nil
}

func (r *TiDBClusterStatusResolver) computeTiFlashStatus(ctx context.Context) (*cloudv1.TiFlashStatus, error) {
	if r.Target.Spec.Cluster.TiFlash.Replicas == 0 {
		// dummy status
		return &cloudv1.TiFlashStatus{Status: cloudv1.ComponentNormal}, nil
	}

	members, conditions, err := r.computeStoreMembers(ctx, &r.Target.Spec.Cluster.TiFlash)
	if err != nil {
		return nil, err
	}
	tiflashNodeScaling, err := r.componentNodeScaling(ctx, cloudv1.ComponentTiFlash)
	if err != nil {
		return nil, err
	}
	var volume *cloudv1.VolumeStatus
	var volModifying bool
	if r.TiDBCluster.Spec.TiFlash != nil {
		volume, volModifying, err = r.componentVolumeStatus(ctx, &r.Target.Spec.Cluster.TiFlash)
		if err != nil {
			return nil, err
		}
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if r.Target.Status.DB != nil && r.Target.Status.DB.TiFlash != nil {
		instanceTypeScalingPhase = r.Target.Status.DB.TiFlash.InstanceTypeScalePhase
	}

	itScaling := instanceTypeScalingFromStoreMembers(members, r.Target.Spec.Cluster.TiFlash.InstanceType) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone

	var status cloudv1.ComponentStatus
	readyStoreCount := 0
	for _, member := range members {
		if member.Status == cloudv1.MemberStatusNormal {
			readyStoreCount++
		}
	}
	switch {
	case readyStoreCount == 0:
		status = cloudv1.ComponentUnavailable
	case r.TiDBCluster.TiFlashUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	case r.TiDBCluster.Status.TiFlash.Phase == v1alpha1.ScalePhase || tiflashNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !r.TiDBCluster.TiFlashAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if r.TiDBCluster.Status.TiFlash.StatefulSet != nil {
		updatedReplicas = r.TiDBCluster.Status.TiFlash.StatefulSet.UpdatedReplicas
	}

	sort.Slice(members, func(i, j int) bool {
		return members[i].Name < members[j].Name
	})

	return &cloudv1.TiFlashStatus{
		Members:                members,
		Status:                 status,
		Volume:                 volume,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             conditions,
	}, nil
}

func (r *TiDBClusterStatusResolver) computeStoreMembers(ctx context.Context, comp cloudv1.ClusterComponent) ([]cloudv1.StoreMember, []metav1.Condition, error) {
	if comp.GetComponentType() != cloudv1.ComponentTiKV && comp.GetComponentType() != cloudv1.ComponentTiFlash {
		return nil, nil, fmt.Errorf("component %s is not store node", comp.GetComponentType())
	}
	if comp.GetReplicas() == 0 {
		return nil, nil, nil
	}
	var members []cloudv1.StoreMember
	instanceSpec, err := r.FindInstanceType(comp.GetInstanceType())
	if err != nil {
		return nil, nil, err
	}
	storePods, err := k8s.SelectPods(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(comp.GetComponentType()))
	if err != nil {
		return nil, nil, err
	}
	storePVCs, err := k8s.SelectPVCs(ctx, r, r.TiDBCluster.Namespace, r.GetComponentPodLabels(comp.GetComponentType()))
	if err != nil {
		return nil, nil, err
	}
	storePVCsMap := make(map[string]corev1.PersistentVolumeClaim)
	for _, pvc := range storePVCs {
		storePVCsMap[pvcNameToPodName(pvc.Name)] = pvc
	}

	storeNodes := make([]corev1.Node, 0)
	for _, p := range storePods {
		if len(p.Spec.NodeName) < 1 {
			continue
		}
		node := corev1.Node{}
		err := r.Client.Get(ctx, client.ObjectKey{
			Name: p.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, nil, err
		}
		storeNodes = append(storeNodes, node)
	}

	nodes, err := r.selectComponentNodes(ctx, comp.GetComponentType(), storeNodes)
	if err != nil {
		return nil, nil, err
	}
	conditions := r.computeComponentConditions(ctx, comp.GetComponentType(), nodes)

	var podToStoreMap map[string]v1alpha1.TiKVStore
	if comp.GetComponentType() == cloudv1.ComponentTiFlash {
		podToStoreMap = podToStore(r.TiDBCluster.Status.TiFlash.Stores, r.TiDBCluster.Status.TiFlash.TombstoneStores)
	} else {
		podToStoreMap = podToStore(r.TiDBCluster.Status.TiKV.Stores, r.TiDBCluster.Status.TiKV.TombstoneStores)
	}

	var previousInstanceTypeMap map[string]string // member-name --> instance-type
	if r.Target.Status.DB != nil {
		if comp.GetComponentType() == cloudv1.ComponentTiFlash && r.Target.Status.DB.TiFlash != nil {
			previousInstanceTypeMap = storeMemberToInstanceType(r.Target.Status.DB.TiFlash.Members)
		} else if r.Target.Status.DB.TiKV != nil {
			previousInstanceTypeMap = storeMemberToInstanceType(r.Target.Status.DB.TiKV.Members)
		}
	}

	for _, pod := range storePods {
		// 1. calculate member status
		store, ok := podToStoreMap[pod.Name]
		var memberStatus cloudv1.MemberStatus
		var upAfterPodCreated bool
		switch {
		case ok && store.State == v1alpha1.TiKVStateUp:
			memberStatus = cloudv1.MemberStatusNormal
			// whether the store is transferred to "Up" after the pod is created (is not an out-of-date "Up" state)
			upAfterPodCreated = store.LastTransitionTime.After(pod.CreationTimestamp.Time) ||
				time.Now().After(pod.CreationTimestamp.Add(podCreationTimeSkew))
		case pod.DeletionTimestamp != nil:
			memberStatus = cloudv1.MemberStatusTerminating
		case ok && store.State == v1alpha1.TiKVStateTombstone && (store.LastTransitionTime.IsZero() || store.LastTransitionTime.After(pod.CreationTimestamp.Time)):
			// a new creating store may have the same name with a previous tombstone store,
			// so we need to check the LastTransitionTime to make sure the store is not out-of-date.
			// `IsZero` is for backward compatibility, ref: https://github.com/pingcap/tidb-operator/pull/4855
			memberStatus = cloudv1.MemberStatusTerminating
		case ok && store.State == v1alpha1.TiKVStateOffline:
			memberStatus = cloudv1.MemberStatusLeaving
		case pod.Status.Phase == corev1.PodPending:
			memberStatus = cloudv1.MemberStatusCreating
		default:
			memberStatus = cloudv1.MemberStatusUnavailable
		}

		// 2. calculate member AZ
		var zone string
		var instanceType = previousInstanceTypeMap[pod.Name]
		for _, node := range storeNodes {
			if node.Name == pod.Spec.NodeName {
				zone = r.GetNodeZoneFromLabels(node.Labels)
				if instanceType == "" || upAfterPodCreated {
					// only change instance-type when the store is transferred to "Up" after re-created for the following case:
					// 1. the pod re-created to a new node with a new instance-type
					// 2. the store keep "Up" state as it's not changed in PD (the pod with new instance-type but old (out-of-date) store info)
					// 3. the store changed to "Disconnected" state in PD
					// 4. the store changed to "Up" state in PD again (the pod with new instance-type and new store info)
					// We can only mark the store is "up-to-date" until step 4, so we delay the instance-type change until step 4.
					instanceType = r.GetNodeInstanceTypeFromLabels(node.Labels)
				}
				break
			}
		}

		if instanceType != "" {
			instanceSpec, err = r.FindInstanceType(instanceType)
			if err != nil {
				return nil, nil, err
			}
		}
		storage, _ := comp.GetStorageSize()
		if pvc, ok := storePVCsMap[pod.Name]; ok {
			pvcStorage := pvc.Spec.Resources.Requests[corev1.ResourceStorage]
			storage = &pvcStorage
		}

		// 3. compose
		members = append(members, cloudv1.StoreMember{
			Name: pod.Name,
			Labels: map[string]string{
				cloudv1.MemberLabelAvailableZone: zone,
			},
			Resources: cloudv1.Resources{
				CPU:     instanceSpec.CPU,
				Memory:  instanceSpec.Memory,
				Storage: storage,
			},
			Status:       memberStatus,
			InstanceType: instanceType,
		})
	}
	return members, conditions, nil
}

func (r *TiDBClusterStatusResolver) componentNodeScaling(ctx context.Context, compType cloudv1.ComponentType) (bool, error) {
	pods, err := k8s.SelectPods(ctx, r.Client, r.TiDBCluster.Namespace, r.GetComponentPodLabels(compType))
	if err != nil {
		return false, err
	}
	scheduledPodCount := 0
	for _, pod := range pods {
		if len(pod.Spec.NodeName) > 0 {
			scheduledPodCount += 1
		}
	}

	podReplicas := 0
	switch compType {
	case cloudv1.ComponentTiDB:
		podReplicas = int(r.TiDBCluster.TiDBStsActualReplicas())
	case cloudv1.ComponentPD:
		podReplicas = int(r.TiDBCluster.PDStsActualReplicas())
	case cloudv1.ComponentTiKV:
		podReplicas = int(r.TiDBCluster.TiKVStsActualReplicas())
	case cloudv1.ComponentTiFlash:
		podReplicas = int(r.TiDBCluster.TiFlashStsActualReplicas())
	case cloudv1.ComponentTiProxy:
		podReplicas = int(r.TiDBCluster.TiProxyStsActualReplicas())
	default:
		// other components does not consider node scaling
		return false, nil
	}

	if scheduledPodCount < podReplicas {
		return true, nil
	}
	return false, nil
}

func (r *TiDBClusterStatusResolver) componentVolumeStatus(ctx context.Context, comp cloudv1.ClusterComponent) (*cloudv1.VolumeStatus, bool /*modifying*/, error) {
	var (
		desiredSizeInTC resource.Quantity
		desiredScInTC   *string // pointer
		dataVolName     v1alpha1.StorageVolumeName
		volumesInTC     map[v1alpha1.StorageVolumeName]*v1alpha1.StorageVolumeStatus
	)

	// collect status
	compType := comp.GetComponentType()
	switch compType {
	case cloudv1.ComponentPD:
		desiredSizeInTC = r.TiDBCluster.Spec.PD.Requests[corev1.ResourceStorage]
		desiredScInTC = r.TiDBCluster.Spec.PD.StorageClassName
		dataVolName = v1alpha1.GetStorageVolumeName("", v1alpha1.PDMemberType)
		volumesInTC = r.TiDBCluster.Status.PD.Volumes
	case cloudv1.ComponentTiKV:
		desiredSizeInTC = r.TiDBCluster.Spec.TiKV.Requests[corev1.ResourceStorage]
		desiredScInTC = r.TiDBCluster.Spec.TiKV.StorageClassName
		dataVolName = v1alpha1.GetStorageVolumeName("", v1alpha1.TiKVMemberType)
		volumesInTC = r.TiDBCluster.Status.TiKV.Volumes
	case cloudv1.ComponentTiFlash:
		if r.TiDBCluster.Spec.TiFlash != nil && len(r.TiDBCluster.Spec.TiFlash.StorageClaims) > 0 {
			desiredSizeInTC = r.TiDBCluster.Spec.TiFlash.StorageClaims[0].Resources.Requests[corev1.ResourceStorage]
			desiredScInTC = r.TiDBCluster.Spec.TiFlash.StorageClaims[0].StorageClassName
		}
		dataVolName = v1alpha1.GetStorageVolumeNameForTiFlash(0)
		volumesInTC = r.TiDBCluster.Status.TiFlash.Volumes
	default:
		r.Log.V(4).Info("No need to compute volume status", "component", compType)
		return nil, false, nil
	}
	if desiredSizeInTC.IsZero() {
		return nil, false, fmt.Errorf("storage size not set in tc for component %s", compType)
	}
	if volumesInTC == nil {
		// operator has not collected the status yet.
		r.Log.V(4).Info("Volume status is empty in tc, so skip collect volume status", "component", compType)
		return nil, false, nil
	}
	dataVolInTC := volumesInTC[dataVolName]
	if dataVolInTC == nil {
		return nil, false, fmt.Errorf("volume status %s not found in tc for component %s", dataVolName, compType)
	}

	// sync volume status
	dataVol := &cloudv1.VolumeStatus{
		BoundCount:           dataVolInTC.BoundCount,
		CurrentCount:         dataVolInTC.CurrentCount,
		ModifiedCount:        dataVolInTC.ModifiedCount,
		CurrentCapacity:      dataVolInTC.CurrentCapacity,
		ModifiedCapacity:     dataVolInTC.ModifiedCapacity,
		CurrentStorageClass:  dataVolInTC.CurrentStorageClass,
		ModifiedStorageClass: dataVolInTC.ModifiedStorageClass,
	}

	// compatible with older versions
	if dataVol.ModifiedCount == 0 {
		dataVol.ModifiedCount = dataVolInTC.ResizedCount
	}
	if dataVol.ModifiedCapacity.IsZero() {
		dataVol.ModifiedCapacity = dataVolInTC.ResizedCapacity
	}
	// remove them after removing the deprecated field
	dataVol.ResizedCount = dataVol.ModifiedCount
	dataVol.ResizedCapacity = dataVol.ModifiedCapacity

	// check if volume is modifying
	modifying := false
	if !comp.GetUseLocalStorage() {
		desiredSize, useSize := comp.GetStorageSize()
		desiredSc, useSc := comp.GetStorageClassName()
		if !useSize || !useSc {
			return nil, false, fmt.Errorf("component %s does not have storage configured", string(compType))
		}
		if desiredSize.Cmp(desiredSizeInTC) != 0 || desiredScInTC == nil || desiredSc != *desiredScInTC ||
			dataVol.CurrentStorageClass == "" { // in non-AWS cloud, the storage class is not set in tc now
			// size/storage-class in cluster is modified, but infra has not updated tc.
			modifying = false
		} else if desiredSize.Cmp(dataVol.CurrentCapacity) != 0 || desiredSc != dataVol.CurrentStorageClass {
			// possible scenarios:
			// 	1. operator has not resized pvc or update volume status. (desiredSize != resizedCapacity)
			// 	2. some pvc are resizing.  (currentCapacity != resizedCapacity)
			//  3. operator has not modified storage class or update volume status. (desiredSc != modifiedStorageClass)
			//  4. some storage classes are modifying. (currentStorageClass != modifiedStorageClass)
			modifying = true
		}
	}

	return dataVol, modifying, nil
}

func instanceTypeScalingFromStoreMembers(members []cloudv1.StoreMember, specInstanceType string) bool {
	for _, member := range members {
		if member.InstanceType != "" && member.InstanceType != specInstanceType {
			return true
		}
	}
	return false
}

func instanceTypeScalingFromPDMembers(members []cloudv1.PDMember, specInstanceType string) bool {
	for _, member := range members {
		if member.InstanceType != "" && member.InstanceType != specInstanceType {
			return true
		}
	}
	return false
}

func instanceTypeScalingFromTiDBMembers(members []cloudv1.TiDBMember, specInstanceType string) bool {
	for _, member := range members {
		if member.InstanceType != "" && member.InstanceType != specInstanceType {
			return true
		}
	}
	return false
}

func podToStore(normals, tombstones map[string]v1alpha1.TiKVStore) map[string]v1alpha1.TiKVStore {
	m := map[string]v1alpha1.TiKVStore{}
	for _, v := range tombstones {
		m[v.PodName] = v
	}
	for _, v := range normals {
		m[v.PodName] = v
	}
	return m
}

func storeMemberToInstanceType(members []cloudv1.StoreMember) map[string]string {
	ret := make(map[string]string, len(members))
	for _, m := range members {
		ret[m.Name] = m.InstanceType
	}
	return ret
}

func tidbMemberToInstanceType(members []cloudv1.TiDBMember) map[string]string {
	ret := make(map[string]string, len(members))
	for _, m := range members {
		ret[m.Name] = m.InstanceType
	}
	return ret
}

func tidbMemberToMemberStatus(members []cloudv1.TiDBMember) map[string]cloudv1.MemberStatus {
	ret := make(map[string]cloudv1.MemberStatus, len(members))
	for _, m := range members {
		ret[m.Name] = m.Status
	}
	return ret
}

func pdMemberToInstanceType(members []cloudv1.PDMember) map[string]string {
	ret := make(map[string]string, len(members))
	for _, m := range members {
		ret[m.Name] = m.InstanceType
	}
	return ret
}

func proxyMemberToInstanceType(members []cloudv1.TiProxyMember) map[string]string {
	ret := make(map[string]string, len(members))
	for _, m := range members {
		ret[m.Name] = m.InstanceType
	}
	return ret
}

// convert a PVC name to the corresponding Pod name,
func pvcNameToPodName(pvcName string) string {
	if idx := strings.LastIndex(pvcName, "-db-"); idx != -1 {

		return pvcName[idx+1:]
	}
	if strings.HasPrefix(pvcName, "db-") {
		return pvcName
	}
	return pvcName
}

func (r *reconcileContext) observeProxyStatus() error {
	if r.networkGardenStatus.ProxyBaseUrl == "" {
		r.Log.V(4).Info("proxy base URL is not ready yet, requeueing...")
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}
	// TODO(aylei): use rule from a lib of proxy instead of hardcode
	clusterProxyBaseUrl := fmt.Sprintf("%s/proxy/cluster/%s", r.networkGardenStatus.ProxyBaseUrl, r.Target.Name)

	return r.ensureStatus(func() error {
		r.Target.Status.Proxy = &cloudv1.ProxyStatus{
			Endpoint: clusterProxyBaseUrl,
		}
		return nil
	})
}

func (r *reconcileContext) observeMetricsStatus() error {
	if r.networkGardenStatus.ProxyBaseUrl == "" {
		r.Log.V(4).Info("proxy base URL is not ready yet, requeueing...")
		return &reconcilerr.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}

	tm := &v1alpha1.TidbMonitor{}
	tmKey := types.NamespacedName{
		Namespace: r.Target.PrefixedId(),
		Name:      consts.ClusterPrefix,
	}
	err := r.shootClient.Get(r.ctx, tmKey, tm)
	if err != nil {
		return errors.Annotate(err, "error observe tidb monitor")
	}
	// cluster metrics are current exposed via the network proxy
	// TODO(aylei): use rule from a lib of proxy instead of hardcode
	metricsBaseUrl := fmt.Sprintf("%s/proxy/cluster/%s/prometheus", r.networkGardenStatus.ProxyBaseUrl, r.Target.Name)

	return r.ensureStatus(func() error {
		r.Target.Status.Metrics = &cloudv1.MetricsStatus{
			BaseURL: metricsBaseUrl,
		}
		return nil
	})
}

func (r *reconcileContext) observeObjectStoreStatus() error {
	if r.networkGardenStatus.BucketName == "" {
		return errors.New("wait for object store bucket of the network")
	}
	err := r.ensureStatus(func() error {
		switch r.ProviderType {
		case consts.ProviderTypeGCP:
			r.Target.Status.ObjectStore = &cloudv1.ObjectStoreStatus{
				GCS: &cloudv1.Path{
					Path: r.networkGardenStatus.BucketName,
				},
			}
		case consts.ProviderTypeAWS:
			r.Target.Status.ObjectStore = &cloudv1.ObjectStoreStatus{
				S3: &cloudv1.Path{
					Path: r.networkGardenStatus.BucketName,
				},
			}
		default:
			return errors.Errorf("unsupported provider type %s", r.ProviderType)
		}
		return nil
	})
	return err
}

// refreshTidbClusterStatus refresh the current status of the tidbcluster resource in reconcileContext
func (r *reconcileContext) refreshTidbClusterStatus() error {
	tc := &v1alpha1.TidbCluster{}
	tcKey := types.NamespacedName{
		Namespace: r.Target.PrefixedId(),
		Name:      consts.ClusterPrefix,
	}
	err := r.shootClient.Get(r.ctx, tcKey, tc)
	if err != nil {
		return errors.Annotate(err, "get tidb cluster")
	}
	r.tc = tc
	return nil
}