package common

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

	"github.com/pingcap/tidb-operator/pkg/apis/label"
	"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/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	apiutil "github.com/tidbcloud/infra-provider/pkg/util"
)

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

// The status reporting to global relies on the reconcile logic of cloudv1.Cluster
// We don't watch the dataplane resources, so the dataplane resources status change
// might take a long time to propagate to global.
// With current controller model, we can reduce the latency either by reconciling
// periodically or by watching dataplane resources. However both have other issues.
// Idealy, we should compute the status on demand when global calls regional cluster-service
func (t *TiDBCluster) resolveStatus(ctx context.Context) error {
	var errs error
	tidb, err := t.computeTiDBStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	pd, err := t.computePDStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tikv, err := t.computeTiKVStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tiflash, err := t.computeTiFlashStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	tiproxy, err := t.computeTiProxyStatus(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	if errs != nil {
		return errs
	}

	// Do not overwrite not own field.
	if t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiDB != nil {
		tidb.AuditReady = t.Cluster.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 t.Cluster.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:         t.Tidbcluster.Spec.Version,
		TiDB:            tidb,
		TiKV:            tikv,
		PD:              pd,
		SecondaryStatus: secondaryStatus,
	}

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

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

	if t.Cluster.Suspending() || t.Cluster.Suspended() {
		phase = cloudv1.ClusterPhaseSuspend
	}

	t.Cluster.Status.Phase = phase
	t.Cluster.Status.Reason = reason
	t.Cluster.Status.DB = dbStatus

	return nil
}
func (t *TiDBCluster) 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 t.Cluster.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 (t *TiDBCluster) 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 t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiDB != nil {
			conditions = t.Cluster.Status.DB.TiDB.Conditions
		}
		component = &t.Cluster.Spec.Cluster.TiDB

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

	if conditions == nil {
		conditions = []metav1.Condition{}
	}
	needSuspend = component.NeedSuspend(t.Cluster.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 := t.Tidbcluster.ComponentIsSuspending(memberType)
	if !needSuspend && !memberSuspending {
		meta.RemoveStatusCondition(&conditions, cloudv1.ComponentSuspended)
		return conditions
	}
	memberSuspended := t.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 (t *TiDBCluster) 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 apiutil.GetAnnotation(t.Cluster, consts.UseSelfManagedNodegroupAnnotation) != consts.AnnoValTrue {
		return nodesOfPod, nil
	}

	nodes, err := k8s.SelectNodes(ctx, t.EnvProvider.GetClient(), t.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 (t *TiDBCluster) computePDStatus(ctx context.Context) (*cloudv1.PDStatus, error) {
	var members []cloudv1.PDMember
	pdSpec := t.Cluster.Spec.Cluster.PD
	instanceSpec, err := t.Provider.FindInstanceType(pdSpec.GetInstanceType())
	if err != nil {
		return nil, err
	}

	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.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, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.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 := t.EnvProvider.GetClient().Get(ctx, client.ObjectKey{
			Name: p.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, err
		}
		nodesOfPod = append(nodesOfPod, node)
	}

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

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

	for name, status := range t.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 = namer.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 = t.Provider.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,
		})
	}

	pdNodeScaling := false
	pdNodeScaling, err = t.componentNodeScaling(ctx, cloudv1.ComponentPD)
	if err != nil {
		return nil, err
	}

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

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if t.Cluster.Status.DB != nil && t.Cluster.Status.DB.PD != nil {
		instanceTypeScalingPhase = t.Cluster.Status.DB.PD.InstanceTypeScalePhase
	}

	// 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 !t.Tidbcluster.PDIsAvailable():
		status = cloudv1.ComponentUnavailable
	case t.Tidbcluster.PDUpgrading() && !itScaling: // we upgrading TidbCluster when scaling for instance type
		status = cloudv1.ComponentUpgrading
	case t.Tidbcluster.PDScaling() || pdNodeScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !t.Tidbcluster.PDAllMembersReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if t.Tidbcluster.Status.PD.StatefulSet != nil {
		updatedReplicas = t.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 (t *TiDBCluster) computeTiDBStatus(ctx context.Context) (*cloudv1.TiDBStatus, error) {
	logger := log.FromContext(ctx)
	// Continue to refresh status if TiDB is suspended
	suspend := t.Tidbcluster.ComponentIsSuspended(v1alpha1.TiDBMemberType)
	// Skip refresh status if status in tc is outdated
	// like STS recreate.
	if !suspend &&
		len(t.Tidbcluster.Status.TiDB.Members) == 0 &&
		t.Cluster.Status.DB != nil &&
		t.Cluster.Status.DB.TiDB != nil {
		return t.Cluster.Status.DB.TiDB, nil
	}

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

	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.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 := t.EnvProvider.GetClient().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, t.EnvProvider.GetClient(), t.GetComponentNodeLabels(cloudv1.ComponentAdminTiDB))
	if err != nil {
		return nil, err
	}
	nodesExtra = append(nodesExtra, nodesOfPod...)

	nodes, err := t.selectComponentNodes(ctx, cloudv1.ComponentTiDB, nodesExtra)
	if err != nil {
		return nil, err
	}
	conditions := t.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 t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiDB != nil {
		previousInstanceTypeMap = tidbMemberToInstanceType(t.Cluster.Status.DB.TiDB.Members)
		previousMemberStatusMap = tidbMemberToMemberStatus(t.Cluster.Status.DB.TiDB.Members)
	}

	for name, status := range t.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.IsPodReady2(pod):
			memberStatus = cloudv1.MemberStatusCreating
		case !status.Health && previousMemberStatusMap[name] == cloudv1.MemberStatusCreating:
			memberStatus = cloudv1.MemberStatusCreating
		case status.Health && k8s.IsPodReady2(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 = namer.GetNodeZoneFromLabels(node.Labels)
				if instanceType == "" || healthAfterPodCreated {
					instanceType = namer.GetNodeInstanceTypeFromLabels(node.Labels)
				}
				break
			}
		}

		if instanceType != "" {
			instanceSpec, err = t.Provider.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 t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiDB != nil {
		instanceTypeScalingPhase = t.Cluster.Status.DB.TiDB.InstanceTypeScalePhase
	}

	itScaling := tidbMembersHaveDisallowedInstanceType(members, t.Cluster.TiDBMemberInstanceTypeAllowed) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone
	tidbNodeCountScaling := t.isNodeScalingCheckEnabled() && t.Cluster.Spec.Cluster.TiDB.Replicas != len(nodes)
	if tidbNodeCountScaling {
		logger.Info("TiDB node scaling", "expected", t.Cluster.Spec.Cluster.TiDB.Replicas, "current", len(nodes))
	}

	var status cloudv1.ComponentStatus
	switch {
	case healthyCount == 0:
		status = cloudv1.ComponentUnavailable
	case t.Tidbcluster.TiDBUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	// FIXME(huaxi): when we scale out tidb, and all tidb pods are running but some are not ready (wait for the readiness gate),
	// we should consider it as scaling instead of unhealthy.
	// Otherwise, the cluster CR's status will be set to Running, but some tidb pods are not ready actually.
	// This added condition will also make the cluster CR's status keep Scaling when scale in for more time.
	case t.Tidbcluster.TiDBScaling() || tidbNodeScaling || tidbNodeCountScaling || itScaling || (t.Cluster.Status.Phase == cloudv1.ClusterPhaseScaling && len(t.Tidbcluster.Status.TiDB.Members) != healthyCount):
		status = cloudv1.ComponentScaling
	case !t.Tidbcluster.TiDBAllMembersReady() || len(t.Tidbcluster.Status.TiDB.Members) != healthyCount:
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if t.Tidbcluster.Status.TiDB.StatefulSet != nil {
		updatedReplicas = t.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(t.Tidbcluster.Spec.TiDB.TokenBasedAuthEnabled, false),
	}, nil
}

func (t *TiDBCluster) computeTiProxyStatus(ctx context.Context) (*cloudv1.TiProxyStatus, error) {
	logger := log.FromContext(ctx)

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

	// compute members and conditions
	var members []cloudv1.TiProxyMember
	var nodesOfPod []corev1.Node
	if t.Tidbcluster.TiProxyStsActualReplicas() != 0 {
		// get pods
		pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.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 := t.EnvProvider.GetClient().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 t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiProxy != nil {
			previousInstanceTypeMap = proxyMemberToInstanceType(t.Cluster.Status.DB.TiProxy.Members)
		}

		// compute member status
		for name, status := range t.Tidbcluster.Status.TiProxy.Members {
			member := cloudv1.TiProxyMember{
				Name:   name,
				Status: cloudv1.MemberStatusUnavailable,
				Labels: map[string]string{},
			}

			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 = namer.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 := t.Provider.FindInstanceType(instanceType)
					if err != nil {
						return nil, err
					}
					member.Resources.CPU = instanceSpec.CPU
					member.Resources.Memory = instanceSpec.Memory
					member.InstanceType = instanceType
				}

				// find zone info
				var zone string
				for _, node := range nodesOfPod {
					if node.Name == pod.Spec.NodeName {
						zone = namer.GetNodeZoneFromLabels(node.Labels)
						break
					}
				}
				member.Labels[cloudv1.MemberLabelAvailableZone] = zone
			}

			members = append(members, member)
		}
	}

	scaling, err := t.componentNodeScaling(ctx, cloudv1.ComponentTiProxy)
	if err != nil {
		return nil, err
	}
	nodes, err := t.selectComponentNodes(ctx, cloudv1.ComponentTiProxy, nodesOfPod)
	if err != nil {
		return nil, err
	}

	var status cloudv1.ComponentStatus

	count := 0
	for _, member := range members {
		if member.Status == cloudv1.MemberStatusNormal {
			count++
		}
	}
	switch {
	case t.Cluster.Spec.Cluster.TiProxy.Replicas != 0 && count == 0:
		status = cloudv1.ComponentUnavailable
	case t.Tidbcluster.Status.TiProxy.Phase == v1alpha1.UpgradePhase && !itScaling:
		status = cloudv1.ComponentUpgrading
	case t.Tidbcluster.Status.TiProxy.Phase == v1alpha1.ScalePhase || scaling || itScaling:
		status = cloudv1.ComponentScaling
	case t.Cluster.Spec.Cluster.TiProxy.Replicas != len(nodes):
		// mark as scaling when the number of nodes is not equal to the number of replicas.
		// For example, scale in/out/up/down TiProxy.
		logger.Info("TiProxy node scaling", "expected", t.Cluster.Spec.Cluster.TiProxy.Replicas, "current", len(nodes))
		status = cloudv1.ComponentScaling
	case count != len(members):
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if t.Tidbcluster.Status.TiProxy.StatefulSet != nil {
		updatedReplicas = t.Tidbcluster.Status.TiProxy.StatefulSet.UpdatedReplicas
	}

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

	conditions := t.computeComponentConditions(ctx, cloudv1.ComponentTiProxy, nodes)

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

func (t *TiDBCluster) computeTiKVStatus(ctx context.Context) (*cloudv1.TiKVStatus, error) {
	logger := log.FromContext(ctx)
	tikvMembers, conditions, err := t.computeStoreMembers(ctx, &t.Cluster.Spec.Cluster.TiKV)
	if err != nil {
		return nil, err
	}
	tikvNodeScaling, err := t.componentNodeScaling(ctx, cloudv1.ComponentTiKV)
	if err != nil {
		return nil, err
	}
	volume, volModifying, err := t.componentVolumeStatus(ctx, &t.Cluster.Spec.Cluster.TiKV)
	if err != nil {
		return nil, err
	}
	additionVolumes, additionVolModifying, err := t.tikvAdditionalVolumesStatus(ctx)
	if err != nil {
		return nil, err
	}
	volModifying = volModifying || additionVolModifying

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiKV != nil {
		instanceTypeScalingPhase = t.Cluster.Status.DB.TiKV.InstanceTypeScalePhase
	}
	nodesOfPod := make([]corev1.Node, 0)
	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.GetComponentPodLabels(cloudv1.ComponentTiKV))
	if err != nil {
		return nil, err
	}
	for _, pod := range pods {
		if len(pod.Spec.NodeName) < 1 {
			continue
		}
		node := corev1.Node{}
		err := t.EnvProvider.GetClient().Get(ctx, client.ObjectKey{
			Name: pod.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, err
		}
		nodesOfPod = append(nodesOfPod, node)
	}

	itScaling := storeMembersHaveDisallowedInstanceType(tikvMembers, t.Cluster.TiKVMemberInstanceTypeAllowed) || instanceTypeScalingPhase != cloudv1.InstanceTypeScalePhaseNone
	nodes, err := t.selectComponentNodes(ctx, cloudv1.ComponentTiKV, nodesOfPod)
	if err != nil {
		return nil, err
	}
	tikvNodeCountScaling := t.isNodeScalingCheckEnabled() && t.Cluster.Spec.Cluster.TiKV.Replicas != len(nodes)
	if tikvNodeCountScaling {
		logger.Info("TiKV node scaling", "expected", t.Cluster.Spec.Cluster.TiKV.Replicas, "current", len(nodes))
	}
	var status cloudv1.ComponentStatus
	switch {
	case !t.Tidbcluster.TiKVIsAvailable():
		status = cloudv1.ComponentUnavailable
	case t.Tidbcluster.TiKVUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	case t.Tidbcluster.TiKVScaling() || tikvNodeScaling || tikvNodeCountScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !t.Tidbcluster.TiKVAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if t.Tidbcluster.Status.TiKV.StatefulSet != nil {
		updatedReplicas = t.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,
		StorageVolumes:         additionVolumes,
		UpdatedReplicas:        updatedReplicas,
		InstanceTypeScalePhase: instanceTypeScalingPhase,
		Conditions:             conditions,
	}, nil
}

func (t *TiDBCluster) computeTiFlashStatus(ctx context.Context) (*cloudv1.TiFlashStatus, error) {
	logger := log.FromContext(ctx)
	if t.Cluster.Spec.Cluster.TiFlash.Replicas == 0 {
		// dummy status
		return &cloudv1.TiFlashStatus{Status: cloudv1.ComponentNormal}, nil
	}

	members, conditions, err := t.computeStoreMembers(ctx, &t.Cluster.Spec.Cluster.TiFlash)
	if err != nil {
		return nil, err
	}
	tiflashNodeScaling, err := t.componentNodeScaling(ctx, cloudv1.ComponentTiFlash)
	if err != nil {
		return nil, err
	}

	var volume *cloudv1.VolumeStatus
	var volModifying bool
	if t.Tidbcluster.Spec.TiFlash != nil {
		volume, volModifying, err = t.componentVolumeStatus(ctx, &t.Cluster.Spec.Cluster.TiFlash)
		if err != nil {
			return nil, err
		}
	}

	var instanceTypeScalingPhase cloudv1.InstanceTypeScalePhase
	if t.Cluster.Status.DB != nil && t.Cluster.Status.DB.TiFlash != nil {
		instanceTypeScalingPhase = t.Cluster.Status.DB.TiFlash.InstanceTypeScalePhase
	}

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

	var status cloudv1.ComponentStatus
	readyStoreCount := 0
	for _, member := range members {
		if member.Status == cloudv1.MemberStatusNormal {
			readyStoreCount++
		}
	}
	nodesOfPod := make([]corev1.Node, 0)
	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.GetComponentPodLabels(cloudv1.ComponentTiFlash))
	if err != nil {
		return nil, err
	}
	for _, pod := range pods {
		if len(pod.Spec.NodeName) < 1 {
			continue
		}
		node := corev1.Node{}
		err := t.EnvProvider.GetClient().Get(ctx, client.ObjectKey{
			Name: pod.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, err
		}
		nodesOfPod = append(nodesOfPod, node)
	}
	nodes, err := t.selectComponentNodes(ctx, cloudv1.ComponentTiFlash, nodesOfPod)
	if err != nil {
		return nil, err
	}
	tiflashNodeCountScaling := t.isNodeScalingCheckEnabled() && t.Cluster.Spec.Cluster.TiFlash.Replicas != len(nodes)
	if tiflashNodeCountScaling {
		logger.Info("TiFlash node scaling", "expected", t.Cluster.Spec.Cluster.TiFlash.Replicas, "current", len(nodes))
	}

	switch {
	case readyStoreCount == 0:
		status = cloudv1.ComponentUnavailable
	case t.Tidbcluster.TiFlashUpgrading() && !itScaling:
		status = cloudv1.ComponentUpgrading
	case t.Tidbcluster.Status.TiFlash.Phase == v1alpha1.ScalePhase || tiflashNodeScaling || tiflashNodeCountScaling || volModifying || itScaling:
		status = cloudv1.ComponentScaling
	case !t.Tidbcluster.TiFlashAllStoresReady():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

	var updatedReplicas int32
	if t.Tidbcluster.Status.TiFlash.StatefulSet != nil {
		updatedReplicas = t.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 (t *TiDBCluster) isNodeScalingCheckEnabled() bool {
	// Node scaling check (replicas != len(nodes)) is only for AWS self-managed nodegroup (e.g. Terminating:Wait).
	return t.Cluster.FeatureEnabled(consts.ClusterFeatureNodeScalingCheck) &&
		apiutil.GetAnnotation(t.Cluster, consts.UseSelfManagedNodegroupAnnotation) == consts.AnnoValTrue
}

func (t *TiDBCluster) 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 := t.Provider.FindInstanceType(comp.GetInstanceType())
	if err != nil {
		return nil, nil, err
	}
	storePods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.GetComponentPodLabels(comp.GetComponentType()))
	if err != nil {
		return nil, nil, err
	}
	storePVCs, err := k8s.SelectPVCs(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.GetComponentPodLabels(comp.GetComponentType()))
	if err != nil {
		return nil, nil, err
	}
	storePVCsMap := make(map[string]corev1.PersistentVolumeClaim)
	for _, pvc := range storePVCs {
		if strings.Contains(pvc.Name, "raftlog") {
			continue
		}
		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 := t.EnvProvider.GetClient().Get(ctx, client.ObjectKey{
			Name: p.Spec.NodeName,
		}, &node)
		if err != nil {
			return nil, nil, err
		}
		storeNodes = append(storeNodes, node)
	}

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

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

	var previousInstanceTypeMap map[string]string // member-name --> instance-type
	if t.Cluster.Status.DB != nil {
		if comp.GetComponentType() == cloudv1.ComponentTiFlash && t.Cluster.Status.DB.TiFlash != nil {
			previousInstanceTypeMap = storeMemberToInstanceType(t.Cluster.Status.DB.TiFlash.Members)
		} else if t.Cluster.Status.DB.TiKV != nil {
			previousInstanceTypeMap = storeMemberToInstanceType(t.Cluster.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 = namer.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 = namer.GetNodeInstanceTypeFromLabels(node.Labels)
				}
				break
			}
		}

		if instanceType != "" {
			instanceSpec, err = t.Provider.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 (t *TiDBCluster) componentNodeScaling(ctx context.Context, compType cloudv1.ComponentType) (bool, error) {
	pods, err := k8s.SelectPods(ctx, t.EnvProvider.GetClient(), t.Tidbcluster.Namespace, t.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(t.Tidbcluster.TiDBStsActualReplicas())
	case cloudv1.ComponentPD:
		podReplicas = int(t.Tidbcluster.PDStsActualReplicas())
	case cloudv1.ComponentTiKV:
		podReplicas = int(t.Tidbcluster.TiKVStsActualReplicas())
	case cloudv1.ComponentTiFlash:
		podReplicas = int(t.Tidbcluster.TiFlashStsActualReplicas())
	case cloudv1.ComponentTiProxy:
		podReplicas = int(t.Tidbcluster.TiProxyStsActualReplicas())
	default:
		// other components does not consider node scaling
		return false, nil
	}

	// In some cases, like scale up + scale out, the cluster CR's status will be Running after the scaling up is finished.
	// To avoid the false positive, we need to check the number of replicas in the cluster CR's spec.
	desiredReplicas := t.Cluster.GetComponent(compType).GetReplicas()
	if scheduledPodCount < podReplicas || podReplicas != desiredReplicas {
		t.Logger.Info("Component is scaling", "component", compType, "scheduledPodCount", scheduledPodCount, "podReplicas", podReplicas, "desiredReplicas", desiredReplicas)
		return true, nil
	}
	return false, nil
}

func (t *TiDBCluster) 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
		desiredVACInTC  *string // pointer
	)

	// collect status
	compType := comp.GetComponentType()
	switch compType {
	case cloudv1.ComponentPD:
		desiredSizeInTC = t.Tidbcluster.Spec.PD.Requests[corev1.ResourceStorage]
		desiredScInTC = t.Tidbcluster.Spec.PD.StorageClassName
		dataVolName = v1alpha1.GetStorageVolumeName("", v1alpha1.PDMemberType)
		volumesInTC = t.Tidbcluster.Status.PD.Volumes
		desiredVACInTC = t.Tidbcluster.Spec.PD.VolumeAttributesClassName
	case cloudv1.ComponentTiKV:
		desiredSizeInTC = t.Tidbcluster.Spec.TiKV.Requests[corev1.ResourceStorage]
		desiredScInTC = t.Tidbcluster.Spec.TiKV.StorageClassName
		dataVolName = v1alpha1.GetStorageVolumeName("", v1alpha1.TiKVMemberType)
		volumesInTC = t.Tidbcluster.Status.TiKV.Volumes
		desiredVACInTC = t.Tidbcluster.Spec.TiKV.VolumeAttributesClassName
	case cloudv1.ComponentTiFlash:
		if t.Tidbcluster.Spec.TiFlash != nil && len(t.Tidbcluster.Spec.TiFlash.StorageClaims) > 0 {
			desiredSizeInTC = t.Tidbcluster.Spec.TiFlash.StorageClaims[0].Resources.Requests[corev1.ResourceStorage]
			desiredScInTC = t.Tidbcluster.Spec.TiFlash.StorageClaims[0].StorageClassName
			desiredVACInTC = t.Tidbcluster.Spec.TiFlash.StorageClaims[0].VolumeAttributesClassName
		}
		dataVolName = v1alpha1.GetStorageVolumeNameForTiFlash(0)
		volumesInTC = t.Tidbcluster.Status.TiFlash.Volumes
	default:
		t.Logger.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 {
		t.Logger.V(4).Info("Volume status is empty in tc", "component", compType)
		return nil, t.gkeVolumeAttributesClassEnabled() && pointer.StringDeref(desiredVACInTC, "") != "", nil
	}
	dataVolInTC := volumesInTC[dataVolName]
	if dataVolInTC == nil {
		return nil, false, fmt.Errorf("volume status %s not found in tc for component %s", dataVolName, compType)
	}

	dataVol := volumeStatusFromOperator(dataVolInTC)

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

		vacEnabled := t.gkeVolumeAttributesClassEnabled() && pointer.StringDeref(desiredVACInTC, "") != ""
		if desiredSize.Cmp(desiredSizeInTC) != 0 || desiredScInTC == nil || desiredSc != *desiredScInTC || (!vacEnabled && 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 vacEnabled {
			vacModifying, err := t.syncGKEVolumeAttributesClassStatus(ctx, compType, dataVolName, desiredSc, *desiredVACInTC, comp.GetReplicas(), dataVol)
			if err != nil {
				return nil, false, err
			}
			modifying = desiredSize.Cmp(dataVol.CurrentCapacity) != 0 || vacModifying
		} 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 (t *TiDBCluster) tikvAdditionalVolumesStatus(ctx context.Context) ([]cloudv1.StorageVolumeStatus, bool /*modifying*/, error) {
	trackVolumeAttributesClass := t.gkeVolumeAttributesClassEnabled()
	if len(t.Cluster.Spec.Cluster.TiKV.StorageVolumes) == 0 {
		// we do not support to remove additional volumes now.
		// so if the additional volumes is empty, we do not need to check the status.
		return nil, false, nil
	}

	desiredVolMap := make(map[string]cloudv1.ComponentStorageVolume)
	for _, vol := range t.Cluster.Spec.Cluster.TiKV.StorageVolumes {
		desiredVolMap[vol.Name] = vol
	}

	var (
		status    []cloudv1.StorageVolumeStatus
		modifying bool
	)
	for _, desiredVolInTC := range t.Tidbcluster.Spec.TiKV.StorageVolumes {
		desiredVol, ok := desiredVolMap[desiredVolInTC.Name]
		if !ok {
			continue
		}
		desiredVAC := pointer.StringDeref(desiredVolInTC.VolumeAttributesClassName, "")

		volName := v1alpha1.GetStorageVolumeName(desiredVolInTC.Name, v1alpha1.TiKVMemberType)
		volInTC := t.Tidbcluster.Status.TiKV.Volumes[volName]
		if volInTC == nil {
			return nil, false, fmt.Errorf("volume status %s not found in tc for component %s", volName, cloudv1.ComponentTiKV)
		}

		volumeStatus := volumeStatusFromOperator(volInTC)

		desiredSC := t.CMEKStorageClassMapper.Name(desiredVol.StorageClassName)
		volumeModifying := desiredVol.StorageSize.Cmp(volInTC.CurrentCapacity) != 0
		if trackVolumeAttributesClass && desiredVAC != "" {
			effectiveDesiredSC := inheritStorageClass(desiredVol.StorageClassName, t.Cluster.Spec.Cluster.TiKV.StorageClassName)
			vacModifying, err := t.syncGKEVolumeAttributesClassStatus(ctx, cloudv1.ComponentTiKV, volName, effectiveDesiredSC, desiredVAC, t.Cluster.Spec.Cluster.TiKV.Replicas, volumeStatus)
			if err != nil {
				return nil, false, err
			}
			volumeModifying = volumeModifying || vacModifying
		} else {
			volumeModifying = volumeModifying || desiredSC != volInTC.CurrentStorageClass
		}

		status = append(status, cloudv1.StorageVolumeStatus{
			Name:         desiredVolInTC.Name,
			VolumeStatus: *volumeStatus,
		})
		modifying = modifying || volumeModifying
		delete(desiredVolMap, desiredVolInTC.Name)
	}
	sort.Slice(status, func(i, j int) bool {
		return status[i].Name < status[j].Name
	})
	if len(desiredVolMap) > 0 {
		// has not update tc yet
		modifying = true
	}
	return status, modifying, nil
}

func inheritStorageClass(storageClass, primaryStorageClass string) string {
	if storageClass == "" {
		return primaryStorageClass
	}
	return storageClass
}

func volumeStatusFromOperator(status *v1alpha1.StorageVolumeStatus) *cloudv1.VolumeStatus {
	volumeStatus := &cloudv1.VolumeStatus{
		BoundCount:           status.BoundCount,
		CurrentCount:         status.CurrentCount,
		ModifiedCount:        status.ModifiedCount,
		CurrentCapacity:      status.CurrentCapacity,
		ModifiedCapacity:     status.ModifiedCapacity,
		CurrentStorageClass:  status.CurrentStorageClass,
		ModifiedStorageClass: status.ModifiedStorageClass,
	}

	// Normalize old and new field names while infra-api exposes both.
	if volumeStatus.ModifiedCount == 0 {
		volumeStatus.ModifiedCount = status.ResizedCount
	}
	if volumeStatus.ModifiedCapacity.IsZero() {
		volumeStatus.ModifiedCapacity = status.ResizedCapacity
	}
	volumeStatus.ResizedCount = volumeStatus.ModifiedCount
	volumeStatus.ResizedCapacity = volumeStatus.ModifiedCapacity
	return volumeStatus
}

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 storeMembersHaveDisallowedInstanceType(members []cloudv1.StoreMember, allowed func(string) bool) bool {
	for _, member := range members {
		if member.InstanceType != "" && !allowed(member.InstanceType) {
			return true
		}
	}
	return false
}

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

func instanceTypeScalingFromTiProxyMembers(members []cloudv1.TiProxyMember, 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
}

func extractPodNameFromPvcName(pvcName string, labels map[string]string) string {
	annPodName := labels[label.AnnPodNameKey]
	if annPodName != "" {
		return annPodName
	}
	return pvcNameToPodName(pvcName)
}

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
}