package common

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

	"github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"gomodules.xyz/jsonpatch/v2"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	schedulingv1 "k8s.io/api/scheduling/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	utilerr "k8s.io/apimachinery/pkg/util/errors"
	"k8s.io/utils/ptr"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	tidbclusterutil "github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/applier"
	"github.com/tidbcloud/infra-provider/pkg/applier/mutatefuncs"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/gkemindomains"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/util/collections"
)

const (
	maxNodeCountPerZone int64 = 100

	// `registry.k8s.io/pause:3.9`
	pauseImage   = "pause"
	pauseVersion = "3.9"

	// this anno mark whether resource labels of nodepool are initialized
	annoKeyResourceLabels             = "gcp.cloud.pingcap.com/resource-labels"
	overProvisioningContainerName     = "over-provisioning"
	overProvisioningPriorityClassName = "over-provisioning-pod"
)

var (
	// InstanceGroup URL: https://www.googleapis.com/compute/{version}/projects/{projectName}/zones/{zone}/instanceGroupManagers/{instanceGroupName}
	instanceGroupUrlRe = regexp.MustCompile(`^https://www.googleapis.com/compute/(\S+)/projects/(\S+)/zones/(\S+)/instanceGroupManagers/(\S+)$`)

	// Parse name from selfLink for request param
	// Full selfLink: https://container.googleapis.com/v1/{name}
	containerSelfLinkToNameRe = regexp.MustCompile(`^https://container.googleapis.com/v1/(\S+)$`)

	defaultNodePoolRequeue = ctrl.Result{RequeueAfter: 10 * time.Second}
)

type NodePool struct {
	*task.BaseTask
	gcpCli gcp.Client
}

func (t *NodePool) Name() string {
	return "NodePool"
}

func NewNodePool(ctx *task.Context, gcpCli gcp.Client) *NodePool {
	return &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}
}

// NOTE: you are recommended to put provide-related logic in Create method, all other methods are provider-agnostic.
func (t *NodePool) Create(ctx context.Context) deploy.Result {
	nps, err := t.desiredNodePools(ctx)
	if err != nil {
		return deploy.Panic(err)
	}

	// Ensure resource labels are set first
	if err := t.setResourceLabels(ctx, nps); err != nil {
		return deploy.Panic(err)
	}

	zoneBootstrapPending := false
	// TiDB Operator v2 does not need this function, because we will set min domains.
	// Otherwise, if we create two tidb clusters with the same instance type and the same AZ, it will be stucked:
	//   - cluster 1: tidb-0(az-0), tidb-1(az-1)
	//   - cluster 2: tidb-0(az-0), tidb-1(az-1)
	// Because there are no nodes in the instance group of az-2,
	// the nodepool's `np.Spec.ForProvider.Autoscaling.MaxNodeCount` will always be 1.
	if !t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		zoneBootstrapPending, err = t.limitNodePoolsForZoneBootstrap(ctx, t.EnvProvider.GetClient(), nps)
		if err != nil {
			return deploy.Panic(err)
		}
	}

	ngApplier := applier.NewApplier(t.Local, mutatefuncs.MutateGKENodePool)
	for _, np := range nps {
		if _, err := ngApplier.Expected(np).Apply(ctx); err != nil {
			return deploy.Error(err)
		}
	}
	deletedUnusedGKEHeterogeneousNodePool, err := t.reconcileGKEComputeClassNodePoolMetadata(ctx)
	if err != nil {
		if k8s.RequeueOnly(err) {
			if requeue := k8s.ShouldRequeue(err); !requeue.IsZero() {
				return deploy.Done().WithRequeuAfter(requeue.RequeueAfter)
			}
		}
		return deploy.Error(err)
	}
	if deletedUnusedGKEHeterogeneousNodePool || zoneBootstrapPending {
		return deploy.Done().WithRequeuAfter(defaultNodePoolRequeue.RequeueAfter)
	}

	return deploy.Done()
}

// setResourceLabels will set resource labels when nodepool is created.
// It will not update resource labels if nodepool exists and has at least 1 resource label
//
// setResourceLabels updates the resource labels for a list of GKE NodePools.
// This method ensures that the NodePools have the correct resource labels applied,
// either by retrieving existing labels or setting new ones.
//
// The method performs the following steps:
// 1. Checks if the NodePool exists locally. If not found:
//   - Sets the `MaxNodeCount` to 0 to avoid node restarts after updating resource labels.
//   - Applies the changes to create the NodePool.
//     2. Verifies the status of the NodePool. If the NodePool is not in a "Running" state,
//     it skips further processing for that NodePool.
//     3. If the resource labels are already recorded in the NodePool's annotations:
//   - Copies them to the target NodePool's annotations.
//     4. If the resource labels are not recorded:
//   - Retrieves them from GCP using the NodePool's self-link.
//   - Initializes them with the default customer resource labels.
//   - Updates the NodePool's resource labels in GCP if it has changed.
//     5. If the resource labels are not empty:
//   - Records the updated labels in the NodePool's annotations.
//
// Parameters:
// - ctx: The context for the operation.
// - nps: A slice of NodePool objects to update.
//
// Returns:
// - An error if any issues occur during the operation, or nil if successful.
func (t *NodePool) setResourceLabels(ctx context.Context, nps []*v1beta1.NodePool) error {
	var errList []error
NPLoop:
	for _, np := range nps {
		var obj v1beta1.NodePool
		key := types.NamespacedName{Name: np.Name}
		if err := t.Local.Get(ctx, key, &obj); err != nil {
			if !apierrors.IsNotFound(err) {
				return err
			}
			// set max node count to 0 if node pool is not found
			// it's for avoiding node restart after updating resource labels
			np.Spec.ForProvider.Autoscaling.MaxNodeCount = ptr.To[int64](0)
			if _, err := applier.NewApplier(t.Local, mutatefuncs.MutateGKENodePool).Expected(np).Apply(ctx); err != nil {
				return err
			}

			errList = append(errList, fmt.Errorf("nodepool %s is not created", np.Name))
			continue
		}
		status := obj.Status.AtProvider.Status
		if status != v1beta1.NodePoolStateRunning {
			errList = append(errList, fmt.Errorf("non-active nodepool %s has the state %q", obj.Name, status))
			continue
		}

		// All existing clusters have resource labels by default, skip them directly
		val, ok := obj.Annotations[annoKeyResourceLabels]
		if ok {
			np.Annotations[annoKeyResourceLabels] = val
			continue
		}

		// Updating resourceLabels after they are running
		// v1beta1.NodeConfig not support resourceLabels, set it manually

		// The name of nodePool must specified in the format
		// 'projects/*/locations/*/clusters/*/nodePools/*'
		nodePoolName, err := t.ParseNameFromSelfLink(obj.Status.AtProvider.SelfLink)
		if err != nil {
			errList = append(errList, fmt.Errorf("cannot parse self link of nodepool %s", obj.Name))
			continue
		}
		ls, err := t.gcpCli.GetNodePoolResourceLabels(ctx, nodePoolName)
		if err != nil {
			errList = append(errList, fmt.Errorf("failed get node pool %s resource labels: %w", obj.Name, err))
			continue
		}

		if ls == nil {
			ls = make(map[string]string)
		}
		labelsChanged := false
		for k, v := range t.resourceLabels() {
			// only set resource labels if they are not set
			if _, ok := ls[k]; !ok {
				ls[k] = v
				labelsChanged = true
			} else if ls[k] != v {
				errList = append(errList, fmt.Errorf("resource label %s of node pool %s is already set to %s, but expected %s", k, obj.Name, ls[k], v))
				continue NPLoop
			}
		}

		// Update the node pool resource labels in GCP if they have changed
		if labelsChanged {
			op, err := t.gcpCli.UpdateNodePoolResourceLabels(ctx, nodePoolName, ls)
			if err != nil {
				errList = append(errList, fmt.Errorf("failed update node pool %s resource labels: %w", obj.Name, err))
				continue
			}
			// An actual update operation has occurred
			if op != nil {
				errList = append(errList, fmt.Errorf("set resource labels for nodepool %s", obj.Name))
				continue
			}
			// no actual update operation, just continue to set the k8s nodepool annotation
		}

		// If the resource labels are not empty, we set them in the k8s nodepool annotation
		if len(ls) != 0 {
			// set anno to avoid GCP api call
			patch := newResourceLabelsPatch(ls)
			if err := k8s.PatchObject(ctx, t.Local, &obj, patch); err != nil {
				errList = append(errList, fmt.Errorf("record resource labels of node pool %s in anno failed: %w", obj.Name, err))
			} else {
				// Skip the current reconciliation after setting the resource labels annotation
				errList = append(errList, fmt.Errorf("set resource labels to annotation of node pool %s", obj.Name))
			}
			continue
		}
	}

	if len(errList) != 0 {
		return utilerr.NewAggregate(errList)
	}

	return nil
}

func newResourceLabelsPatch(ls map[string]string) k8s.ObjectPatch {
	var kvs []string
	for k, v := range ls {
		kvs = append(kvs, k+":"+v)
	}
	sort.Strings(kvs)

	return func(obj client.Object) *jsonpatch.Operation {
		anno, op := obj.GetAnnotations(), "replace"
		if anno == nil || anno[annoKeyResourceLabels] == "" {
			op = "add"
		}
		return &jsonpatch.Operation{
			Operation: op,
			Path:      `/metadata/annotations/` + strings.ReplaceAll(annoKeyResourceLabels, "/", "~1"),
			Value:     strings.Join(kvs, ","),
		}
	}
}

func (t *NodePool) PostCreate(ctx context.Context) deploy.Result {
	remoteCli := t.EnvProvider.GetClient()

	zones := t.Zones()
	zoneCount := len(zones)

	if err := t.createStandbyNodeOPDeployments(ctx, remoteCli, zones); err != nil {
		return deploy.Error(fmt.Errorf("create standby node OP deployments: %w", err))
	}

	// TODO(ideascf): remove this. It's actually not needed/used.
	// check pods scheduled when creating the cluster, and then mark `NodeProvisioned` to true.
	// but do not check pods scheduled when scaling the cluster,
	// because the pods may even not be created
	// as the `replicas` in TidbCluster may lazily update until nodes are ready for scaling up/down.
	// NOTE: PD `Scaling` is not treated as Cluster `Scaling` phase now.
	if (t.Cluster.Status.Phase != cloudv1.ClusterPhaseScaling && t.Cluster.Status.Phase != cloudv1.ClusterPhaseRunning) &&
		(t.Cluster.Status.DB == nil || t.Cluster.Status.DB.PD == nil || t.Cluster.Status.DB.PD.Status != cloudv1.ComponentScaling) {
		if err := t.isPodsScheduled(ctx, remoteCli); err != nil {
			return deploy.Error(err)
		}

		t.CommonK8sClusterStatus.NodeProvisioned = ptr.To(true)

		// if instance type scale up/down required,
		// the cluster phase will be updated to "Scaling" in the previous round of reconcile when observing TidbCluster.
		// so we only do the following operations when the cluster phase is "Scaling" to reduce performance overhead.
		return deploy.Done()
	}

	// Note: all code you write here will be executed in every PostCreate phase, even if the cluster is not in Scaling, Creating phase.
	// So, please make sure the code is idempotent and has no side effects.
	wait := false
	for _, comp := range t.Cluster.GetAllComponentsV2() {
		// scale-up stage 1: create pods for overprovisioning if needed
		// create dedicated overprovisioning pods for each component so that they can be done one by one,
		// and not all blocking in Scaling phase until all finished.
		if err := t.createOverProvisioningPods(ctx, remoteCli, comp, zoneCount); err != nil {
			return deploy.Error(err)
		}

		// scale-up stage 2: wait for overprovisioning pods to be scheduled (e.g. wait for nodes to be ready)
		// It's used for creating nodes with target instance type in the dataplane.
		if ready, err := t.waitOverProvisioningPodsReady(ctx, remoteCli, comp); err != nil {
			return deploy.Error(err)
		} else if !ready {
			wait = true
			continue
		}

		// scale-up stage 3: restart pods to the new nodes
		// if fact, we only record components in this task, and let the TidbCluster task to do the real restart
		if err := t.recordCompsForRestart(ctx, comp); err != nil {
			return deploy.Error(err)
		}

		// scale-up stage 4: delete overprovisioning pods after the corresponding component pods are ready
		if ready, err := t.waitAndDeleteOverProvisioningPods(ctx, remoteCli, comp); err != nil {
			return deploy.Error(err)
		} else if !ready {
			wait = true
			continue
		}

		// scale-up stage 5: wait overprovisioning pods deleted and trans phase to `None`
		if ready, err := t.waitOverProvisioningPodsDeleted(ctx, remoteCli, comp); err != nil {
			return deploy.Error(err)
		} else if !ready {
			wait = true
			continue
		}
	}
	if wait {
		return deploy.Error(k8s.NewRetriableError(k8s.RequeueOnlyError, defaultNodePoolRequeue))
	}

	// scale-up stage 6: clear the recorded components in annotation after scaling finished.
	if err := t.clearRecordedComps(ctx); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

// isPodsScheduled return nil if all pods are scheduled
func (t *NodePool) isPodsScheduled(ctx context.Context, remoteCli client.Client) error {
	for _, c := range t.Cluster.GetAllComponentSpecs() {
		if c.GetReplicas() < 1 {
			continue
		}

		podList := &corev1.PodList{}
		if err := remoteCli.List(ctx, podList,
			client.InNamespace(t.Namer.GetTiDBNamespace()),
			machingLabels4ComponentPods(t.Namer.GetTiDBClusterName(), c.GetComponentType()),
		); err != nil {
			return err
		}
		scheduledPodCount := 0
		for _, pod := range podList.Items {
			if len(pod.Spec.NodeName) > 0 {
				scheduledPodCount += 1
			}
		}

		if scheduledPodCount < c.GetReplicas() {
			return fmt.Errorf("wait %s pod scheduled, scheduled: %d, replicas: %d",
				c.GetComponentType(), scheduledPodCount, c.GetReplicas())
		}
	}
	return nil
}

// Delete deletes overprovisioning pods only. Ordinary shared NodePools are
// retained until the network is deleted; API-managed heterogeneous NodePools
// may be reclaimed separately by the idle GC in the create reconcile.
func (t *NodePool) Delete(ctx context.Context) deploy.Result {
	remoteCli := t.EnvProvider.GetClient()

	// Delete all overprovisioning pods when deleting the cluster
	if err := t.deleteOverProvisioningDeployments(ctx, remoteCli, nil); err != nil {
		return deploy.Error(err)
	}

	if err := t.deleteStandbyNodeOPDeployments(ctx, remoteCli); err != nil {
		return deploy.Error(fmt.Errorf("delete standby node OP deployments: %w", err))
	}
	return deploy.Done()
}

func (t *NodePool) PostDelete(_ context.Context) deploy.Result {
	return deploy.Done()
}

// desiredNodePools generate nodepools expected to createOrUpdate
func (t *NodePool) desiredNodePools(_ context.Context) ([]*v1beta1.NodePool, error) {
	zones := t.Zones()
	if len(zones) < 1 {
		return nil, nil
	}

	sa := t.GetGCPNodePoolServiceAccountEmail()

	// NodePools are shared between clusters and components
	sharedNamer := t.Namer.WithClusterName(common.ClustersShared).
		WithComponent(common.ComponentShared).
		WithTiDBNamespace(common.ClustersShared)

	npsMap := make(map[string]*v1beta1.NodePool)
	for _, c := range t.Cluster.GetAllComponentSpecs() {
		// TiFlash can be not specified, and the `comp` would be an empty struct
		// Just continue, cleanup will be done separately if it is in the case of scale in to 0
		if c.GetReplicas() < 1 || t.isComponentExpectedToPause(c) {
			continue
		}

		it := c.GetInstanceType()

		np := &v1beta1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name: nodepool.GetGKENodePoolName(t.Cluster, it, t.GetNodeVersion(false)),
				// TODO: should not use namer.Labels to nodepool
				// All of above labels are useless...
				Labels:      sharedNamer.Labels(),
				Annotations: map[string]string{},
			},
			Spec: v1beta1.NodePoolSpec{
				ForProvider: v1beta1.NodePoolParameters{
					Cluster: t.GetGCPClusterRef(),
					Autoscaling: &v1beta1.NodePoolAutoscaling{
						Enabled:      ptr.To(true),
						MaxNodeCount: ptr.To(maxNodeCountPerZone),
						MinNodeCount: ptr.To(int64(0)),
					},
					Config: &v1beta1.NodeConfig{
						Labels:         t.kubernetesLabels(t.GetNodeVersion(true)),
						DiskSizeGb:     ptr.To(int64(100)),
						DiskType:       ptr.To(utils.GetGcpNodepoolDiskType(it)),
						ImageType:      ptr.To("cos_containerd"),
						MachineType:    &it,
						ServiceAccount: &sa, // node sa
						OauthScopes: []string{
							"https://www.googleapis.com/auth/cloud-platform",
						},
						Taints: []*v1beta1.NodeTaint{
							{
								Key:    "dedicated",
								Value:  "tidb-" + common.ComponentShared,
								Effect: "NO_SCHEDULE",
							},
						},
					},
					InitialNodeCount: ptr.To(int64(0)),
					Locations:        zones,
					Management: &v1beta1.NodeManagementSpec{
						// Do it manually
						AutoUpgrade: ptr.To(false),
						AutoRepair:  ptr.To(false),
					},
				},
				ResourceSpec: xpv1.ResourceSpec{
					ProviderConfigReference: &xpv1.Reference{
						Name: t.GetGCPProviderConfigName(),
					},
				},
			},
		}

		if t.Cluster.GetProviderType() == cloudv1.ProviderTypeGKE && t.Cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
			np.Spec.ForProvider.Config.Labels[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue
			np.Spec.ForProvider.Config.Taints = append(np.Spec.ForProvider.Config.Taints, taintToNodeTaint(utils.TaintWith3LayerArch()))
			// add project and tenant label to nodepool labels
			np.Labels[cloudv1.TagLabelPrefix+common.Project] = t.Cluster.GetProjectID()
			np.Labels[cloudv1.TagLabelPrefix+common.Tenant] = t.Cluster.GetTenantID()
		}

		// GKE adds this taint to ARM node pools. Declare it in the desired spec so
		// provider-gcp does not repeatedly treat the GKE-managed taint as drift.
		if getInstanceArch(t.Provider, it) == consts.ARM64ArchName {
			np.Spec.ForProvider.Config.Taints = append(np.Spec.ForProvider.Config.Taints, taintToNodeTaint(utils.TaintWithArch(consts.ARM64ArchName)))
		}

		req := desiredNodePoolRequest{
			nodePoolName:  np.Name,
			instanceType:  c.GetInstanceType(),
			componentType: c.GetComponentType(),
		}

		// check against with the existing nodepool
		old, ok := npsMap[req.nodePoolName]
		if ok {
			if err := t.mergeDesiredNodePool(old, np, req); err != nil {
				return nil, err
			}
			continue
		}

		npsMap[req.nodePoolName] = np
	}

	nps := make([]*v1beta1.NodePool, 0)
	for _, np := range npsMap {
		nps = append(nps, np)
	}
	return nps, nil
}

func taintToNodeTaint(t *corev1.Taint) *v1beta1.NodeTaint {
	nt := &v1beta1.NodeTaint{
		Key:   t.Key,
		Value: t.Value,
	}
	switch t.Effect {
	case corev1.TaintEffectNoExecute:
		nt.Effect = "NO_EXECUTE"
	case corev1.TaintEffectNoSchedule:
		nt.Effect = "NO_SCHEDULE"
	case corev1.TaintEffectPreferNoSchedule:
		nt.Effect = "PREFER_NO_SCHEDULE"
	default:
		nt.Effect = "EFFECT_UNSPECIFIED"
	}
	return nt
}

func (t *NodePool) GetGCPClusterRef() string {
	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", t.EnvProvider.GetAccount(), t.EnvProvider.GetRegion(), t.EnvProvider.GetClusterName())
}

func (t *NodePool) isComponentExpectedToPause(c cloudv1.ClusterComponent) bool {
	return c.NeedSuspend(t.Cluster.Spec.Cluster.SuspendAction)
}

// resourceLabels return the labels apply to gce instances
// The key must only contain lowercase letters ([a-z]), numeric characters ([0-9]), underscores (_) and dashes (-), and must start with a letter.
func (t *NodePool) resourceLabels() map[string]string {
	l := utils.GenerateCommonNodePoolLabels(t.Cluster)
	return cloudv1.TagsFromLabels(l)
}

func (t *NodePool) kubernetesLabels(k8sVersion string) map[string]string {
	l := t.resourceLabels()

	disableNodePoolDataPlaneLabel := t.Cluster.FeatureEnabled(consts.ClusterFeatureDisableNodePoolDataPlaneLabel)
	if !disableNodePoolDataPlaneLabel {
		// NodePool type, in order to make the node selector of tidb components
		// Or it may be scheduled to admin nodepool when the machine type is the same
		l[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane
	}

	// Add k8s version label
	if k8sVersion != "" {
		l[consts.K8sVersionLabelKey] = k8sVersion
	}

	return l
}

// limitNodePoolsForZoneBootstrap preserves the one-node-per-zone bootstrap only
// for active consumers that do not yet enforce minDomains.
func (t *NodePool) limitNodePoolsForZoneBootstrap(
	ctx context.Context,
	remoteCli client.Client,
	nps []*v1beta1.NodePool,
) (bool, error) {
	var tidbClusters *opv1alpha1.TidbClusterList
	var deployments *appsv1.DeploymentList
	limited := false

	for _, np := range nps {
		if np.Spec.ForProvider.Autoscaling == nil ||
			np.Spec.ForProvider.Autoscaling.MaxNodeCount == nil ||
			*np.Spec.ForProvider.Autoscaling.MaxNodeCount <= 1 {
			continue
		}

		missingZone, err := t.nodePoolHasMissingZone(ctx, np)
		if err != nil {
			return false, err
		}
		if !missingZone {
			continue
		}

		needsBootstrap := !gkemindomains.Applicable(t.Cluster)
		if !needsBootstrap {
			if tidbClusters == nil {
				tidbClusters = &opv1alpha1.TidbClusterList{}
				if err := remoteCli.List(ctx, tidbClusters); err != nil {
					return false, fmt.Errorf("list TiDBClusters: %w", err)
				}

				deployments = &appsv1.DeploymentList{}
				if err := remoteCli.List(ctx, deployments, client.MatchingLabels{
					consts.ComponentLabelKey: consts.OverProvisioningComponentLabelValue,
				}); err != nil {
					return false, fmt.Errorf("list over-provisioning Deployments: %w", err)
				}
			}
			needsBootstrap = t.nodePoolHasConsumerWithoutMinDomains(np, tidbClusters.Items, deployments.Items)
		}
		if needsBootstrap {
			np.Spec.ForProvider.Autoscaling.MaxNodeCount = ptr.To(int64(1))
			limited = true
		}
	}
	return limited, nil
}

func (t *NodePool) nodePoolHasMissingZone(ctx context.Context, np *v1beta1.NodePool) (bool, error) {
	current := &v1beta1.NodePool{}
	if err := t.Local.Get(ctx, types.NamespacedName{Name: np.Name}, current); err != nil {
		return false, err
	}

	instanceGroups := make(map[string][]string, len(current.Status.AtProvider.InstanceGroupUrls))
	for _, url := range current.Status.AtProvider.InstanceGroupUrls {
		parts := instanceGroupUrlRe.FindStringSubmatch(url)
		if len(parts) != 5 {
			return false, fmt.Errorf("parse instance group url error, find %d submatch strings", len(parts))
		}
		instanceGroups[parts[3]] = parts
	}

	for _, zone := range np.Spec.ForProvider.Locations {
		instanceGroup, ok := instanceGroups[zone]
		if !ok {
			return true, nil
		}
		size, err := t.gcpCli.GetInstanceGroupSize(ctx, instanceGroup[2], zone, instanceGroup[4])
		if err != nil {
			return false, fmt.Errorf("failed get instance group size error: %w", err)
		}
		if size < 1 {
			return true, nil
		}
	}
	return false, nil
}

func (t *NodePool) nodePoolHasConsumerWithoutMinDomains(
	np *v1beta1.NodePool,
	tidbClusters []opv1alpha1.TidbCluster,
	deployments []appsv1.Deployment,
) bool {
	componentTypes := []cloudv1.ComponentType{
		cloudv1.ComponentPD,
		cloudv1.ComponentTiKV,
		cloudv1.ComponentTiDB,
		cloudv1.ComponentTiProxy,
		cloudv1.ComponentTiFlash,
	}
	for i := range tidbClusters {
		tc := &tidbClusters[i]
		getter := tidbclusterutil.NewV1Getter(tc)
		for _, componentType := range componentTypes {
			if getter.ComponentDesiredReplicas(componentType) > 0 &&
				t.nodeSelectorMatchesNodePool(np, getter.ComponentNodeSelector(componentType)) &&
				!gkemindomains.HasMinDomainsConstraint(tc, componentType) {
				return true
			}
		}
	}

	for i := range deployments {
		deployment := &deployments[i]
		if deployment.Spec.Replicas == nil || *deployment.Spec.Replicas <= 0 ||
			!t.nodeSelectorMatchesNodePool(np, deployment.Spec.Template.Spec.NodeSelector) {
			continue
		}

		hasMinDomains := false
		for _, constraint := range deployment.Spec.Template.Spec.TopologySpreadConstraints {
			if constraint.TopologyKey == corev1.LabelTopologyZone &&
				constraint.MaxSkew == 1 &&
				constraint.MinDomains != nil &&
				*constraint.MinDomains == int32(gkemindomains.RequiredZoneCount) &&
				constraint.WhenUnsatisfiable == corev1.DoNotSchedule {
				hasMinDomains = true
				break
			}
		}
		if !hasMinDomains {
			return true
		}
	}
	return false
}

func (t *NodePool) nodeSelectorMatchesNodePool(np *v1beta1.NodePool, nodeSelector map[string]string) bool {
	if np.Spec.ForProvider.Config == nil || np.Spec.ForProvider.Config.MachineType == nil ||
		nodeSelector[consts.InstanceTypeLabelKey] != *np.Spec.ForProvider.Config.MachineType {
		return false
	}

	k8sVersion := nodeSelector[consts.K8sVersionLabelKey]
	return k8sVersion == "" || nodepool.GetGKENodePoolName(t.Cluster, *np.Spec.ForProvider.Config.MachineType, k8sVersion) == np.Name
}

func (t *NodePool) ParseNameFromSelfLink(selfLink string) (name string, err error) {
	strs := containerSelfLinkToNameRe.FindStringSubmatch(selfLink)
	if len(strs) != 2 {
		return "", fmt.Errorf("parse nodepool name from selflink error, parsed strings: %q", strs)
	}
	return strs[1], nil
}

func (t *NodePool) createOverProvisioningPods(ctx context.Context, remoteCli client.Client, comp cloudv1.ClusterComp, zoneCount int) error {
	logger := log.FromContext(ctx).WithValues("component", comp.GetComponentType(), "group", comp.GetGroupId())
	var replicas int

	n := t.GetNamer(ctx, comp)
	tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
	if err != nil {
		return err
	}
	compType := comp.GetComponentType()
	compSpec := comp.(cloudv1.ClusterComponentSpecWrapped)
	compStatus := comp.(cloudv1.ClusterComponentStatusWrapped)
	scaleUp, err := nodepool.IsComponentScalingUp(t.Provider, compSpec, compStatus)
	if scaleUp && err == nil && nodepool.IsComponentScalingIn(compSpec, compStatus) {
		// If the component is scaling up and in at the same time,
		// we should calculate opReplicas based on the previous replicas.
		// For example, if we scale up and in (6 -> 3) tikv, we should create 6 overprovisioning pods.
		// Since scale up is processed first, we can get the previous replicas from the tc's spec.
		replicas = int(tcGetter.ComponentReplicas(compType))
	} else {
		replicas = comp.GetReplicas()
	}

	if replicas < 1 || t.isComponentExpectedToPause(comp) {
		return nil
	}

	if compStatus.GetInstanceTypeScalePhase() != cloudv1.InstanceTypeScalePhaseNone {
		// only create overprovisioning pods when the instance type scale phase is not "None"
		return nil
	}

	podList := &corev1.PodList{}
	if err := remoteCli.List(ctx, podList,
		client.InNamespace(n.GetTiDBNamespace()),
		machingLabels4ComponentPods(n.GetTiDBClusterName(), compType),
	); err != nil {
		return err
	}

	var (
		currentIt           string
		currentNodeSelector map[string]string
	)
	for _, pod := range podList.Items {
		if len(currentNodeSelector) == 0 && len(pod.Spec.NodeSelector) > 0 {
			currentNodeSelector = pod.Spec.NodeSelector
		}
		if it := pod.Spec.NodeSelector[consts.InstanceTypeLabelKey]; it != "" {
			// in `None` phase, all pods should have the same instance type
			currentIt = it
			currentNodeSelector = pod.Spec.NodeSelector
			break
		}
	}
	targetIt := comp.GetInstanceType()
	usedDisableGKEHeterogeneousFallback := false
	if currentIt == "" {
		if len(podList.Items) == 0 {
			return nil // no pod is scheduled, this should not happen when scaling up/down in `None` phase
		}
		nodeSelector, originalIt, ok, err := t.resolveDisableGKEHeterogeneousOPNodeSelector(comp, currentNodeSelector)
		if err != nil {
			return err
		}
		if !ok {
			return nil
		}
		currentNodeSelector = nodeSelector
		currentIt = originalIt
		usedDisableGKEHeterogeneousFallback = true
	}

	if currentIt == targetIt {
		return nil // no need to create over-provisioning pods
	}

	useMinDomains := gkemindomains.Applicable(t.Cluster)
	targetInstanceType, err := t.Provider.FindInstanceType(targetIt)
	if err != nil {
		return fmt.Errorf("failed query arch of instance type(%s): %w", targetIt, err)
	}

	opReplicas := int32(replicas)
	if replicas%zoneCount != 0 {
		// always over provisioning zoneCount*N nodes and let extra nodes to be deleted by autoscaler,
		// then we don't need to calculate replicas by AZs.
		opReplicas = int32(replicas/zoneCount*zoneCount + zoneCount)
	}
	logger.Info("need to create over-provisioning pods",
		"current-instance-type", currentIt, "target-instance-type", targetIt,
		"component-replicas", replicas, "over-provisioning-replicas", opReplicas,
		"disable-gke-heterogeneous-fallback", usedDisableGKEHeterogeneousFallback,
		"gke-v1-min-domains", useMinDomains,
	)

	// ensure the nodeSelector is consistent with the corresponding tc component pod, except `namer.InstanceTypeLabelKey`.
	nodeSelector := collections.DeepCopyMap(currentNodeSelector)
	nodeSelector[consts.InstanceTypeLabelKey] = targetIt
	// NOTE: in tidb operator v2 we add different zones into pods' node selector
	// TODO(liubo02): maybe refactor it to avoid overprovision evenly spread issue
	delete(nodeSelector, consts.ZoneLabelKey)
	// namer.BuildOverProvisioningLabelValue(targetIt)
	opName := namer.BuildOverProvisioningNameV2(compType, comp.GetGroupId(), targetIt)
	opDeploy := buildOpDeployment(&buildOpDeploymentParams{
		OpName:       opName,
		OpReplicas:   opReplicas,
		OpUsedFor:    consts.OverProvisioningUsedForTidbcluster,
		Comp:         comp,
		NodeSelector: nodeSelector,
		Namer:        t.GetNamer(ctx, comp),
		Cluster:      t.Cluster,
		Zones:        t.Zones(),
		Arch:         targetInstanceType.Arch,
	})
	if useMinDomains {
		if opDeploy.Annotations == nil {
			opDeploy.Annotations = make(map[string]string)
		}
		opDeploy.Annotations[annoGKEV1MinDomainsInstanceTypeRollout] = consts.AnnoValTrue
	}

	// Existing Deployments own their in-progress rollout.
	if err := remoteCli.Create(ctx, opDeploy); err != nil && !apierrors.IsAlreadyExists(err) {
		return err
	}

	// if over-provisioning pods are created, the previous scaling phase should be `None`
	comp.TransInstanceTypeScalePhase(ctx, cloudv1.InstanceTypeScalePhaseCreatingNodes)

	return nil
}

type buildOpDeploymentParams struct {
	OpName       string
	OpReplicas   int32
	OpUsedFor    string
	Comp         cloudv1.ClusterComponentSpecWrapped
	NodeSelector map[string]string

	Namer   *namer.Namer
	Cluster *cloudv1.Cluster
	Zones   []string
	Arch    string
}

func buildOpDeployment(params *buildOpDeploymentParams) *appsv1.Deployment {
	compType := params.Comp.GetComponentType()
	otherCompTypes := make([]string, 0, 3)
	for _, compTypes2 := range params.Cluster.GetAllComponentTypes() {
		if compType != compTypes2 {
			otherCompTypes = append(otherCompTypes, string(compTypes2))
		}
	}

	opDeploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			// each component has its own over-provisioning deployment
			Name:      params.OpName,
			Namespace: params.Namer.GetTiDBNamespace(),
			Labels:    buildOpDeploymentLabels(params.Namer, params.Comp, params.OpUsedFor),
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: ptr.To(params.OpReplicas),
			Selector: &metav1.LabelSelector{
				MatchLabels: buildOpDeploymentLabels(params.Namer, params.Comp, params.OpUsedFor),
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: buildOpDeploymentLabels(params.Namer, params.Comp, params.OpUsedFor),
				},
				Spec: corev1.PodSpec{
					TerminationGracePeriodSeconds: ptr.To(int64(0)),
					NodeSelector:                  params.NodeSelector,
					Affinity: &corev1.Affinity{
						PodAntiAffinity: &corev1.PodAntiAffinity{
							RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
								{
									// anti-affinity with other over-provisioning pods for all components
									NamespaceSelector: &metav1.LabelSelector{}, // `An empty selector ({}) matches all namespaces`
									LabelSelector: &metav1.LabelSelector{
										MatchLabels: map[string]string{
											consts.InstanceLabelKey:  params.Namer.GetTiDBClusterName(), // always `db`
											consts.ComponentLabelKey: consts.OverProvisioningComponentLabelValue,
										},
									},
									TopologyKey: corev1.LabelHostname,
								},
								{
									// anti-affinity with all TiDB Cluster components in other namespaces
									NamespaceSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{
											{
												Key:      corev1.LabelMetadataName, // kubernetes.io/metadata.name
												Operator: metav1.LabelSelectorOpNotIn,
												Values:   []string{params.Namer.GetTiDBNamespace()},
											},
										},
									},
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{
											{
												Key:      consts.ComponentLabelKey,
												Operator: metav1.LabelSelectorOpIn,
												Values: []string{
													string(cloudv1.ComponentPD),
													string(cloudv1.ComponentTiKV),
													string(cloudv1.ComponentTiDB),
													string(cloudv1.ComponentTiFlash),
													string(cloudv1.ComponentTiProxy),
												},
											},
										},
									},
									TopologyKey: corev1.LabelHostname,
								},
								{
									// anti-affinity with other TiDB Cluster components in the same namespace
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{
											{
												Key:      consts.ComponentLabelKey,
												Operator: metav1.LabelSelectorOpIn,
												Values:   otherCompTypes,
											},
										},
									},
									TopologyKey: corev1.LabelHostname,
								},
								{
									// anti-affinity with other tidb node groups
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{
											{
												Key:      consts.InstanceLabelKey,
												Operator: metav1.LabelSelectorOpNotIn,
												Values:   []string{params.Namer.GetTiDBClusterName()},
											},
										},
									},
									TopologyKey: corev1.LabelHostname,
								},
							},
						},
					},
					TopologySpreadConstraints: []corev1.TopologySpreadConstraint{
						{
							// spread over-provisioning pods evenly across all zones,
							// in fact all zones should have the same number of over-provisioning pods as `replicas` is `zoneCount*N`
							MaxSkew:           1,
							MinDomains:        ptr.To(int32(len(params.Zones))),
							TopologyKey:       corev1.LabelTopologyZone,
							WhenUnsatisfiable: corev1.DoNotSchedule,
							LabelSelector: &metav1.LabelSelector{
								MatchLabels: buildOpDeploymentLabels(params.Namer, params.Comp, params.OpUsedFor),
							},
						},
					},
					Tolerations: utils.TolerationWithComponentShared(),
					Containers: []corev1.Container{
						{
							Name:  overProvisioningContainerName,
							Image: config.GetThirdPartyImage(pauseImage, pauseVersion),
							Resources: corev1.ResourceRequirements{
								Requests: corev1.ResourceList{
									corev1.ResourceCPU: resource.MustParse("5m"),
								},
							},
						},
					},
				},
			},
		},
	}

	if params.Cluster.GetProviderType() == cloudv1.ProviderTypeGKE {
		if params.Cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
			opDeploy.Spec.Template.Spec.Tolerations = append(opDeploy.Spec.Template.Spec.Tolerations, *utils.TolerationWith3LayerArch())
			opDeploy.Spec.Template.Spec.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue
		}

		if params.Arch == consts.ARM64ArchName {
			opDeploy.Spec.Template.Spec.Tolerations = append(opDeploy.Spec.Template.Spec.Tolerations, *utils.TolerationWithArch(params.Arch))
		}
	}
	disableNodePoolDataPlaneLabel := params.Cluster.FeatureEnabled(consts.ClusterFeatureDisableNodePoolDataPlaneLabel)
	if !disableNodePoolDataPlaneLabel {
		opDeploy.Spec.Template.Spec.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane
	}

	return opDeploy
}

func buildOpDeploymentLabels(namer *namer.Namer, comp cloudv1.ClusterComponentSpecWrapped, opUsedFor string) map[string]string {
	return map[string]string{
		consts.InstanceLabelKey:                  namer.GetTiDBClusterName(),
		consts.ComponentLabelKey:                 consts.OverProvisioningComponentLabelValue,
		consts.OverProvisioningComponentLabelKey: string(comp.GetComponentType()),
		consts.LabelKeyTiDBGroupId:               comp.GetGroupId(),
		consts.OverProvisioningUsedForLabelKey:   opUsedFor,
	}
}

func (t *NodePool) waitOverProvisioningPodsReady(ctx context.Context, remoteCli client.Client, comp cloudv1.ClusterComp) (ready bool, _ error) {
	logger := log.FromContext(ctx).WithValues("component", comp.GetComponentType(), "group", comp.GetGroupId())
	n := t.GetNamer(ctx, comp)

	// only check over-provisioning pods are ready if scaling phase is `CreatingNodes`
	if comp.GetInstanceTypeScalePhase() != cloudv1.InstanceTypeScalePhaseCreatingNodes {
		return true, nil
	}

	opDeployList := &appsv1.DeploymentList{}
	if err := remoteCli.List(ctx, opDeployList,
		client.InNamespace(n.GetTiDBNamespace()),
		machingLabels4TidbClusterOPDeployment(comp),
	); err != nil {
		return false, err
	}
	if len(opDeployList.Items) == 0 {
		logger.Info("no over-provisioning deployment found, requeue")
		return false, nil
	}

	for _, opDeploy := range opDeployList.Items {
		if !namer.HasOverProvisioningNamePrefixV2(opDeploy.Name) {
			continue
		}
		if opDeploy.Status.ReadyReplicas != *opDeploy.Spec.Replicas {
			logger.Info("over-provisioning deployment is not ready, requeue",
				"deployment", opDeploy.Name, "ready", opDeploy.Status.ReadyReplicas, "desired", *opDeploy.Spec.Replicas)
			return false, nil
		} else {
			logger.Info("over-provisioning deployment is ready",
				"deployment", opDeploy.Name, "ready", opDeploy.Status.ReadyReplicas, "desired", *opDeploy.Spec.Replicas)
		}
	}

	// transfer the scaling phase to `CordoningNodes`,
	// but in fact there is no need to cordon nodes as we use nodeSelector to schedule pods to the new nodes
	comp.TransInstanceTypeScalePhase(ctx, cloudv1.InstanceTypeScalePhaseCordoningNodes)

	return true, nil
}

func (t *NodePool) createOPPriorityClass(ctx context.Context, remoteCli client.Client) error {
	preemptionPolicy := corev1.PreemptNever
	pc := &schedulingv1.PriorityClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: overProvisioningPriorityClassName,
		},
		// set priority and preemptionPolicy to ensure over provisioning pod can be preempted by TC pods and can't preempt others.
		// Note:
		//  The default priority is 0 and default preemptionPolicy is PreemptLowerPriority.
		//  Follow https://github.com/kubernetes/autoscaler/blob/master/cluster-autoscaler/FAQ.md#how-does-cluster-autoscaler-work-with-pod-priority-and-preemption,
		//  the pod having priority less than -10 can't trigger node scale out, therefore we set it between (-10, 0), and now is set to -5.
		Value:            -5,
		Description:      "This PriorityClass is used for over provisioning pods. Pods with this PriorityClass means it's used for provision nodes in advance, and they can be preempted.",
		PreemptionPolicy: &preemptionPolicy,
	}
	return k8s.CreateIfNotExist(ctx, remoteCli, pc)
}

// createStandbyNodeOPDeployments creates over provisioning deployments used for creating standby nodes
func (t *NodePool) createStandbyNodeOPDeployments(ctx context.Context, remoteCli client.Client, zones []string) error {
	logger := log.FromContext(ctx)

	deploymentList := &appsv1.DeploymentList{}
	if err := remoteCli.List(ctx, deploymentList,
		client.InNamespace(t.Cluster.PrefixedId()),
		machingLabels4StandbyOPDeployments()); err != nil {
		return err
	}
	name2Deployment := map[string]appsv1.Deployment{}
	for _, each := range deploymentList.Items {
		name2Deployment[each.Name] = each
	}

	if err := t.createOPPriorityClass(ctx, remoteCli); err != nil {
		return err
	}
	for _, comp := range t.Cluster.GetAllComponentSpecs() {
		instanceType := comp.GetInstanceType()
		compType := comp.GetComponentType()
		it, err := t.Provider.FindInstanceType(instanceType)
		if err != nil {
			return fmt.Errorf("failed query arch of instance type(%s): %w", instanceType, err)
		}
		for _, zone := range zones {
			tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
			if err != nil {
				return err
			}
			opDeployName := namer.BuildOverProvisioningNameForStandbyNodeV2(compType, comp.GetGroupId(), zone)
			cnt := t.Cluster.GetStandbyNodeCnt(compType, zone)
			if cnt == 0 {
				deployment, ok := name2Deployment[opDeployName]
				if !ok {
					continue
				}

				logger.Info("delete OP deployment", "component", compType, "name", opDeployName)
				if err := remoteCli.Delete(ctx, &deployment); err != nil {
					return fmt.Errorf("delete OP deployment(%s) for standby node: %w", opDeployName, err)
				}
				continue
			}
			nodeSelector := tcGetter.ComponentNodeSelector(compType)
			if nodeSelector == nil {
				logger.Info("component is not deployed in TC or its node selectors are not set", "component", compType)
				continue
			}
			nodeSelector[consts.InstanceTypeLabelKey] = instanceType
			nodeSelector[corev1.LabelTopologyZone] = zone // create standby node's opDeployment for each zone
			opDeployment := buildOpDeployment(&buildOpDeploymentParams{
				OpName:       opDeployName,
				OpReplicas:   int32(cnt),
				OpUsedFor:    consts.OverProvisioningUsedForStandbyNode,
				Comp:         comp,
				NodeSelector: nodeSelector,
				Namer:        t.GetNamer(ctx, comp),
				Cluster:      t.Cluster,
				Zones:        zones,
				Arch:         it.Arch,
			})
			// set resource requests to ensure one node only can schedule one of either an over provisioning pod of standby node or a TC pod
			for idx, each := range opDeployment.Spec.Template.Spec.Containers {
				if each.Name != overProvisioningContainerName {
					continue
				}

				resources, err := t.ConfigProvider.GetResourceSpec(t.Provider, t.Cluster.GetComponent(compType))
				if err != nil {
					return fmt.Errorf("get resources for %s for standby pod: %w", compType, err)
				}
				requests := resources.Resources.Requests.DeepCopy()
				delete(requests, corev1.ResourceStorage)
				logger.Info("set resource requests for OP deployment", "requests", requests)
				opDeployment.Spec.Template.Spec.Containers[idx].Resources.Requests = requests
			}

			opDeployment.Spec.Template.Spec.PriorityClassName = overProvisioningPriorityClassName
			logger.Info("apply OP deployment for standby node", "name", opDeployName, "cnt", cnt)
			if err := k8s.Apply(ctx, remoteCli, opDeployment); err != nil {
				return fmt.Errorf("create over provision deployment(%s) for standby node: %w", opDeployName, err)
			}
		}
	}
	return nil
}

// deleteStandbyNodeOPDeployments cleans all over provisioning deployments used for creating standby nodes
func (t *NodePool) deleteStandbyNodeOPDeployments(ctx context.Context, remoteCli client.Client) error {
	deploymentList := &appsv1.DeploymentList{}
	if err := remoteCli.List(ctx, deploymentList,
		client.InNamespace(t.Cluster.PrefixedId()),
		machingLabels4StandbyOPDeployments()); err != nil {
		return err
	}

	for _, each := range deploymentList.Items {
		if err := remoteCli.Delete(ctx, &each); client.IgnoreNotFound(err) != nil {
			return fmt.Errorf("delete over provision deployment(%s): %w", each.Name, err)
		}
	}
	return nil
}

func (t *NodePool) recordCompsForRestart(ctx context.Context, comp cloudv1.ClusterComp) error {
	tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
	if err != nil {
		return err
	}

	phase := comp.GetInstanceTypeScalePhase()
	if phase == cloudv1.InstanceTypeScalePhaseCordoningNodes || phase == cloudv1.InstanceTypeScalePhaseRestartingPods {
		// record components if scaling phase is `CordoningNodes` or `RestartingPods`
		t.Cluster.AppendInstanceTypeChangedComponent(comp, tcGetter.Name())
	} else { // TODO(ideascf): I think we should only remove the component when phase is `None`
		t.Cluster.RemoveInstanceTypeChangedComponent(comp, tcGetter.Name())
	}

	if comp.GetInstanceTypeScalePhase() == cloudv1.InstanceTypeScalePhaseCordoningNodes {
		// transfer the scaling phase to `RestartingPods`
		comp.TransInstanceTypeScalePhase(ctx, cloudv1.InstanceTypeScalePhaseRestartingPods)
		t.Logger.Info("TidbCluster is needed to rolling-update to target nodegroup", "component", comp)
	}

	return nil
}

func (t *NodePool) waitAndDeleteOverProvisioningPods(ctx context.Context, remoteCli client.Client, comp cloudv1.ClusterComp) (ready bool, _ error) {
	tcGetter, err := t.GetTCGetter(ctx, comp.GetGroupId())
	if err != nil {
		return false, err
	}
	namer := t.GetNamer(ctx, comp)
	compType := comp.GetComponentType()
	if comp.GetInstanceTypeScalePhase() != cloudv1.InstanceTypeScalePhaseRestartingPods {
		// only check pods ready if scaling phase is `RestartingPods`
		return true, nil
	}

	// check all pods re-scheduled to the new nodes and ready
	componentInstanceType := tcGetter.ComponentNodeSelector(compType)[consts.InstanceTypeLabelKey]
	componentReady := tcGetter.ComponentAllInstancesReady(compType)
	if componentInstanceType != comp.GetInstanceType() {
		// wait for the instance type update to be triggered
		// NOTE: a reconcile should be triggered when setting `nodeSelector` to the target instance type
		t.Logger.Info("Waiting for the instance type update to be triggered",
			"component", compType, "current-instance-type", componentInstanceType, "target-instance-type", comp.GetInstanceType())
		return false, nil
	}
	if !componentReady {
		t.Logger.Info("Waiting for the component to be ready before deleting over provisioning pods", "component", compType)
		return false, nil
	}

	// check any pod has been updated to the target instance type to avoid no-resync happens in tidb-operator (and `componentReady` is out of date)
	podList := &corev1.PodList{}
	if err := remoteCli.List(ctx, podList,
		client.InNamespace(namer.GetTiDBNamespace()),
		machingLabels4ComponentPods(namer.GetTiDBClusterName(), compType),
	); err != nil {
		return false, err
	}

	for _, pod := range podList.Items {
		if it := pod.Spec.NodeSelector[consts.InstanceTypeLabelKey]; it != comp.GetInstanceType() {
			t.Logger.Info("Waiting for the pods to be updated to the target instance type before deleting over provisioning pods",
				"component", compType, "current-instance-type", it, "target-instance-type", comp.GetInstanceType())
			// the cluster phase should not be changed if the pods is updating to the target instance type
			return false, nil
		}
	}

	// delete over-provisioning deployment for this component
	if err := t.deleteOverProvisioningDeployments(ctx, remoteCli, comp); err != nil {
		return false, err
	}

	// transfer the scaling phase to `DeletingNodes`,
	// but in fact we only delete the over-provisioning deployments, and let the nodes be deleted by the autoscaler
	comp.TransInstanceTypeScalePhase(ctx, cloudv1.InstanceTypeScalePhaseDeletingNodes)

	return true, nil
}

func (t *NodePool) waitOverProvisioningPodsDeleted(ctx context.Context, remoteCli client.Client, comp cloudv1.ClusterComp) (ready bool, _ error) {
	logger := log.FromContext(ctx)
	if comp.GetInstanceTypeScalePhase() != cloudv1.InstanceTypeScalePhaseDeletingNodes {
		// only check over-provisioning pods are deleted if scaling phase is `DeletingNodes`
		return true, nil
	}

	// check over-provisioning pods(deployment) for this component are deleted
	opDeployList := &appsv1.DeploymentList{}
	if err := remoteCli.List(ctx, opDeployList, client.InNamespace(t.Cluster.PrefixedId()), machingLabels4TidbClusterOPDeployment(comp)); err != nil {
		return false, err
	}
	for _, opDeploy := range opDeployList.Items {
		if !namer.HasOverProvisioningNamePrefixV2(opDeploy.Name) {
			logger.Info("deployment is labeled OverProvisioning but its name is without prefix 'op-'", "deployment", opDeploy.Name)
			continue
		}
		t.Logger.Info("Waiting for the over-provisioning pods to be deleted", "deployment", opDeploy.Name)
		return false, nil
	}

	comp.TransInstanceTypeScalePhase(ctx, cloudv1.InstanceTypeScalePhaseNone)
	return true, nil
}

func (t *NodePool) deleteOverProvisioningDeployments(ctx context.Context, remoteCli client.Client, comp cloudv1.ClusterComp) error {
	logger := log.FromContext(ctx)
	opDeployList := &appsv1.DeploymentList{}
	if err := remoteCli.List(ctx, opDeployList, client.InNamespace(t.Cluster.PrefixedId()), machingLabels4TidbClusterOPDeployment(comp)); err != nil {
		return err
	}
	for _, opDeploy := range opDeployList.Items {
		if !namer.HasOverProvisioningNamePrefixV2(opDeploy.Name) {
			logger.Info("deployment is labeled OverProvisioning but its name is without prefix 'op-'", "deployment", opDeploy.Name)
			continue
		}
		if err := remoteCli.Delete(ctx, &opDeploy); client.IgnoreNotFound(err) != nil {
			return err
		}
	}

	return nil
}

func (t *NodePool) clearRecordedComps(ctx context.Context) error {
	for groupId, comps := range t.Cluster.AllComponentsV2GroupByGroupId() {
		allDone := true
		for _, comp := range comps {
			phase := comp.GetInstanceTypeScalePhase()
			if phase != cloudv1.InstanceTypeScalePhaseNone {
				allDone = false
			}
		}
		if !allDone {
			continue
		}

		// delete previous annotation if no update needed so that it will not affect the next reconcile triggered by another instance type scale
		tcGetter, err := t.GetTCGetter(ctx, groupId)
		if err != nil {
			return err
		}
		t.Cluster.ClearInstanceTypeChangedComponents(tcGetter.Name())
	}

	return nil
}

func machingLabels4TidbClusterOPDeployment(comp cloudv1.ClusterComponentSpecWrapped) client.MatchingLabels {
	mLabels := client.MatchingLabels{
		consts.ComponentLabelKey:               consts.OverProvisioningComponentLabelValue,
		consts.OverProvisioningUsedForLabelKey: consts.OverProvisioningUsedForTidbcluster,
	}
	if comp != nil {
		mLabels[consts.LabelKeyTiDBGroupId] = comp.GetGroupId()
		mLabels[consts.OverProvisioningComponentLabelKey] = string(comp.GetComponentType())
	}

	return mLabels
}

func machingLabels4StandbyOPDeployments() client.MatchingLabels {
	return client.MatchingLabels(map[string]string{
		consts.OverProvisioningUsedForLabelKey: consts.OverProvisioningUsedForStandbyNode,
	})
}

func machingLabels4ComponentPods(tcName string, compType cloudv1.ComponentType) client.MatchingLabels {
	return client.MatchingLabels(map[string]string{
		consts.ComponentLabelKey: string(compType),
		consts.InstanceLabelKey:  tcName,
	})
}