package cluster

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

	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"

	"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"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	observeAfter = 1 * 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
}

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

	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)
	}
	if errs != nil {
		return errs
	}

	var phase cloudv1.ClusterPhase
	secondaryStatus := cloudv1.DBSecondaryStatusHealthy
	// PD partial failure cannot be felt by user, so the status is hidden
	if 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 is unavailable
	case tidb.Status == cloudv1.ComponentUnavailable || pd.Status == cloudv1.ComponentUnavailable || tikv.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:
		phase = cloudv1.ClusterPhaseUpgrading
	case tidb.Status == cloudv1.ComponentScaling || tikv.Status == cloudv1.ComponentScaling || tiflash.Status == cloudv1.ComponentScaling:
		phase = cloudv1.ClusterPhaseScaling
	default:
		phase = cloudv1.ClusterPhaseRunning
	}
	reason := fmt.Sprintf("pd: %s, tidb: %s, tikv: %s, tiflash: %s", pd.Status, tidb.Status, tikv.Status, tiflash.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.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
	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
}
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
	}
	nodes, err := k8s.SelectNodes(ctx, r.Client, r.GetComponentNodeLabels(cloudv1.ComponentPD))
	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)
		}

		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 nodes {
					if node.Name == pod.Spec.NodeName {
						instanceType = r.GetNodeInstanceTypeFromLabels(node.Labels)
						break
					}
				}
			}
		}

		members = append(members, cloudv1.PDMember{
			Name: name,
			Resources: cloudv1.Resources{
				CPU:     instanceSpec.CPU, // show new CPU/Memory when scaling for instance type now similar with `Storage`
				Memory:  instanceSpec.Memory,
				Storage: &pdSpec.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 := instanceTypeScalingFromPDMembers(members, pdSpec.GetInstanceType()) || 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
	}

	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

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

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

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

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

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

		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():
		status = cloudv1.ComponentUnhealthy
	default:
		status = cloudv1.ComponentNormal
	}

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

	return &cloudv1.TiDBStatus{
		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
	}

	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
	}

	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
	}
	storeNodes, err := k8s.SelectNodes(ctx, r.Client, r.GetComponentNodeLabels(comp.GetComponentType()))
	if err != nil {
		return nil, nil, err
	}
	conditions := r.computeComponentConditions(ctx, comp.GetComponentType(), storeNodes)

	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)
		case pod.DeletionTimestamp != nil:
			memberStatus = cloudv1.MemberStatusTerminating
		case ok && store.State == v1alpha1.TiKVStateTombstone:
			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
			}
		}

		// 3. compose
		storage, _ := comp.GetStorageSize()
		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) {
	nodes, err := k8s.SelectNodes(ctx, r.Client, r.GetComponentNodeLabels(compType))
	if err != nil {
		return false, err
	}
	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())
	default:
		// other components does not consider node scaling
		return false, nil
	}
	// if nodes number > component replicas, there might be failover or node with delete protection that need manual
	// intervention, we handle such case internally and does not expose the issue to our client
	if len(nodes) < 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
		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]
		dataVolName = v1alpha1.GetStorageVolumeName("", v1alpha1.PDMemberType)
		volumesInTC = r.TiDBCluster.Status.PD.Volumes
	case cloudv1.ComponentTiKV:
		desiredSizeInTC = r.TiDBCluster.Spec.TiKV.Requests[corev1.ResourceStorage]
		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]
		}
		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, use := comp.GetStorageSize()
		if !use {
			return nil, false, fmt.Errorf("component %s does not have storage configured", string(compType))
		}
		if desiredSize.Cmp(desiredSizeInTC) != 0 {
			// size in cluster is modified, but infra has not updated tc.
			modifying = false
		} else if desiredSize.Cmp(dataVol.CurrentCapacity) != 0 {
			// possible scenarios:
			// 	1. operator has not resized pvc or update volume status. (desiredSize != resizedCapacity)
			// 	2. some pvc are resizing.  (currentCapacity != resizedCapacity)
			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 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 (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
}