package common

import (
	"context"
	"fmt"
	"regexp"
	"strconv"
	"time"

	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/errors"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"
	"golang.org/x/mod/semver"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/audit"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task/common/statementlog"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/provider"
	"github.com/tidbcloud/infra-provider/pkg/util"

	nodepool "github.com/tidbcloud/infra-provider/pkg/nodepool"
)

const (
	// https://docs.pingcap.com/tidb/stable/tidb-distributed-execution-framework#usage
	minDistTaskSupportedVersion = "v8.1"
)

type TiDBGroup struct {
	*task.BaseTask
}

func NewTiDBGroup(ctx *task.Context) deploy.Task {
	return &TiDBGroup{
		BaseTask: &task.BaseTask{Context: ctx},
	}
}

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

func (t *TiDBGroup) buildClusterContext(extraGroup *cloudv1.ExtraGroupCluster) (cc *ClusterContext, err error) {
	cc = &ClusterContext{
		Target:         t.Cluster,
		Provider:       &t.Config,
		ConfigProvider: t.ConfigProvider,
		// Injected providers
		Cluster: t.newTidbGroupLabelProvider(extraGroup),
	}

	return enableKMS(t.BaseTask, cc)
}

func enableKMS(t *task.BaseTask, cc *ClusterContext) (*ClusterContext, error) {
	isEnable, ok := t.Cluster.Annotations[consts.KMSEnabledKey]
	if ok && isEnable == consts.AnnoValTrue {
		keyARN := t.KMSGetter.Get()
		if keyARN == "" {
			return nil, fmt.Errorf("kms is enabled for %s/%s, but the kms key is not found", t.Cluster.Namespace, t.Cluster.Name)
		}

		keyRegion, keyId, err := util.ParseKeyARN(keyARN)
		if err != nil {
			return nil, util.WrapError("parse master key arn", err)
		}
		cc.KmsRegion = &keyRegion
		cc.KmsKeyId = &keyId
		if t.Config.Provider == cloudv1.ProviderTypeGKE {
			// TODO: This should be deleted once confirmed no cluster on GKS uses TiKV KMS encryption
			cc.MountKmsSecret = true
		}
	}
	return cc, nil
}

func (t *TiDBGroup) Create(ctx context.Context) deploy.Result {
	logger := log.FromContext(ctx)

	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		logger.Info("skipping TiDBGroup creation for TiDB Operator V2",
			"reason", "TiDB Operator V2 support not implemented")
		// TODO(csuzhangxc): TiDB Operator V2 support
		return deploy.Done()
	}

	if len(t.Cluster.Spec.ExtraGroups) == 0 {
		logger.Info("no extra groups to create")
		return deploy.Done()
	}

	extraGroups := t.Cluster.GetExtraGroups()
	logger.Info("creating TiDB extra groups",
		"totalExtraGroups", len(extraGroups),
		"clusterPhase", t.Cluster.Status.Phase)

	var errs *multierror.Error
	for _, extraGroup := range extraGroups {
		if extraGroup.Deleted() {
			logger.Info("cleaning up deleted extra group",
				"extraGroup", extraGroup.Name,
				"tidbReplicas", extraGroup.Spec.TiDB.Replicas,
				"tiproxyReplicas", extraGroup.Spec.TiProxy.Replicas)
			errs = multierror.Append(errs, t.ensureZeroReplicasExtraGroupCleaned(ctx, &extraGroup))
			continue
		}

		logger.Info("reconciling extra TiDB cluster",
			"extraGroup", extraGroup.Name,
			"tidbReplicas", extraGroup.Spec.TiDB.Replicas,
			"tidbInstanceType", extraGroup.Spec.TiDB.InstanceType,
			"tiproxyReplicas", extraGroup.Spec.TiProxy.Replicas,
			"tiproxyInstanceType", extraGroup.Spec.TiProxy.InstanceType)
		errs = multierror.Append(errs, t.ReconcileExtraTiDBCluster(ctx, &extraGroup))
	}
	return deploy.ErrorOrDone(errs.ErrorOrNil())
}

func (t *TiDBGroup) PostCreate(ctx context.Context) deploy.Result {
	logger := log.FromContext(ctx)

	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		logger.Info("skipping TiDBGroup PostCreate for TiDB Operator V2",
			"reason", "TiDB Operator V2 support not implemented")
		// TODO(csuzhangxc): TiDB Operator V2 support
		return deploy.Done()
	}

	if len(t.Cluster.Spec.ExtraGroups) == 0 {
		logger.Info("no extra groups to observe in PostCreate")
		return deploy.Done()
	}

	cli := t.EnvProvider.GetClient()
	extraGroups := t.Cluster.GetExtraGroups()

	logger.Info("observing TiDB extra groups in PostCreate",
		"totalExtraGroups", len(extraGroups),
		"clusterPhase", t.Cluster.Status.Phase)

	var errs *multierror.Error

	for _, extraGroup := range extraGroups {
		if extraGroup.Deleted() {
			logger.Info("skipping observation of deleted extra group",
				"extraGroup", extraGroup.Name)
			continue
		}

		logger.Info("observing extra TiDB cluster",
			"extraGroup", extraGroup.Name,
			"tidbReplicas", extraGroup.Spec.TiDB.Replicas,
			"tiproxyReplicas", extraGroup.Spec.TiProxy.Replicas)
		errs = multierror.Append(errs, t.observeExtraTidbCluster(ctx, cli, &extraGroup))
	}

	return deploy.ErrorOrDone(errs.ErrorOrNil())
}

func (t *TiDBGroup) Delete(ctx context.Context) deploy.Result {
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
		// TODO(csuzhangxc): TiDB Operator V2 support
		return deploy.Done()
	}

	if len(t.Cluster.Spec.ExtraGroups) == 0 {
		return deploy.Done()
	}

	extraGroups := t.Cluster.GetExtraGroups()

	var errs *multierror.Error
	for _, extraGroup := range extraGroups {
		errs = multierror.Append(errs, t.ensureZeroReplicasExtraGroupCleaned(ctx, &extraGroup))
	}
	return deploy.ErrorOrDone(errs.ErrorOrNil())
}

// Disabled function list:
// PD, TiKV
// Pod ENI
// TiFlash TLS
// Bootstrap SQL
// mutateTiKVForDedicatedRaftEngineStorage
// elasticHTAP

func (t *TiDBGroup) ReconcileExtraTiDBCluster(ctx context.Context, extraGroup *cloudv1.ExtraGroupCluster) error {
	logger := log.FromContext(ctx)
	t.BeginAt = util.Pointer(time.Now())

	logger.Info("starting extra TiDB cluster reconciliation",
		"extraGroup", extraGroup.Name,
		"clusterPhase", t.Cluster.Status.Phase,
		"tidbReplicas", extraGroup.Spec.TiDB.Replicas,
		"tidbInstanceType", extraGroup.Spec.TiDB.InstanceType,
		"tiproxyReplicas", extraGroup.Spec.TiProxy.Replicas)

	tidbMountOOMPathForNew := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeOOMPath)
	tidbMountOOMPathForExisting := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeOOMPath)
	tidbMountOOMPathForUpgrading := tidbMountOOMPathForNew && t.Cluster.Status.Phase == cloudv1.ClusterPhaseUpgrading
	tidbMountReplayerPathForNew := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeReplayerPath)
	tidbMountReplayerPathForExisting := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeReplayerPath)
	tidbMountReplayerPathForUpgrading := tidbMountReplayerPathForNew && t.Cluster.Status.Phase == cloudv1.ClusterPhaseUpgrading

	logger.Info("feature flags for extra TiDB cluster",
		"extraGroup", extraGroup.Name,
		"tidbMountOOMPathForNew", tidbMountOOMPathForNew,
		"tidbMountOOMPathForExisting", tidbMountOOMPathForExisting,
		"tidbMountReplayerPathForNew", tidbMountReplayerPathForNew,
		"tidbMountReplayerPathForExisting", tidbMountReplayerPathForExisting)

	cli := t.EnvProvider.GetClient()

	logger.Info("building cluster context for extra group",
		"extraGroup", extraGroup.Name)
	c, err := t.buildClusterContext(extraGroup)
	if err != nil {
		logger.Error(err, "failed to build cluster context",
			"extraGroup", extraGroup.Name)
		return err
	}

	logger.Info("building TiDB cluster spec for extra group",
		"extraGroup", extraGroup.Name)
	tc, err := t.buildTidbCluster(c, extraGroup)
	if err != nil {
		logger.Error(err, "failed to build TiDB cluster",
			"extraGroup", extraGroup.Name)
		return err
	}

	extraGroupNamer := t.TidbGroupNamer(extraGroup.Spec)

	tc.Spec.Cluster = &opv1alpha1.TidbClusterRef{
		// it's the default access group TiDB Cluster name
		Name: extraGroupNamer.GetDefaultTiDBName(),
	}
	tc.Spec.TiFlash = nil
	tc.Spec.PD = nil
	tc.Spec.TiKV = nil
	tc.Name = extraGroupNamer.GetTiDBName()

	logger.Info("configured TiDB cluster spec for extra group",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"defaultTiDBName", extraGroupNamer.GetDefaultTiDBName(),
		"namespace", extraGroupNamer.GetTiDBNamespace())

	// As we have fixed `tidbGroupLabelProvider` to let it returns the correct labels for this `extraGroup`,
	// we can safely use the labels from `tidbGroupLabelProvider` without any overlay here.
	//
	// An example labels for tidb:
	// gardener.wg.prefix.name: 10592080259617370235-group1-tidb
	// tidb-component: tidb
	// tidb-namespace: tidb10592080259617370235
	// tidbcloud-cluster: 10592080259617370235-group1
	//
	// controllers/cluster/task/eks/extragroup/extra_self_managed_nodegroup.go:561 ng.Labels = util.ConcatenateTags(ng.Labels, ngNamer.Labels())
	if !t.useClusterAutoscaler() {
		// it's used for EKS self-managed nodegroup, if cluster is managed in GKE and AKS with cluster autoscaler, it's not needed.
		tc.Spec.TiDB.NodeSelector = extraGroupNamer.WithComponent(string(cloudv1.ComponentTiDB)).Labels()
		logger.Info("using EKS self-managed nodegroup labels for TiDB",
			"extraGroup", extraGroup.Name,
			"nodeSelector", tc.Spec.TiDB.NodeSelector)
	} else {
		// set for GKE and AKS node labels
		tc.Spec.TiDB.NodeSelector[consts.InstanceTypeLabelKey] = extraGroup.Spec.TiDB.InstanceType
		logger.Info("using cluster autoscaler instance type for TiDB",
			"extraGroup", extraGroup.Name,
			"instanceType", extraGroup.Spec.TiDB.InstanceType,
			"provider", t.Config.Provider)
	}
	tc.Spec.TiDB.Replicas = int32(extraGroup.Spec.TiDB.Replicas)

	logger.Info("set TiDB replicas for extra group",
		"extraGroup", extraGroup.Name,
		"replicas", tc.Spec.TiDB.Replicas)

	// Distributed Execution Framework
	if distTaskSupported(tc.Spec.Version) {
		// avoid invalid name
		validName := regexp.MustCompile(`^[a-zA-Z0-9_-]+$`).MatchString
		if !validName(extraGroup.Name) {
			return fmt.Errorf("invalid extraGroup.Name: %s", extraGroup.Name)
		}

		tc.Spec.TiDB.Arguments = append(tc.Spec.TiDB.Arguments, fmt.Sprintf("--tidb-service-scope=%s", extraGroup.Name))
	}

	// CMEK
	if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseCMEK) {
		// set the storage class for the tidb cluster object
		err = t.useCMEKStorageClass(tc)
		if err != nil {
			return err
		}
	}

	// set sem in tc configuration, because it's has been set in tidb config
	// extragroup is created after cluster is ready.
	err = setSEMforTiDBGroup(ctx, t.Cluster, cli, tc)
	if err != nil {
		return err
	}

	// Mutate TidbCluster
	obj := &opv1alpha1.TidbCluster{}
	obj.Annotations = map[string]string{}
	obj.Namespace = extraGroupNamer.GetTiDBNamespace()
	obj.Name = extraGroupNamer.GetTiDBName()

	logger.Info("creating or updating TiDB cluster",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", obj.Name,
		"namespace", obj.Namespace)

	opResult, err := createOrUpdateTidbClusterWithRetry(ctx, cli, obj, func() error {
		current := obj.DeepCopy()
		defer func() {
			diff := cmp.Diff(current, obj, cmpopts.EquateEmpty())
			if diff == "" {
				logger.Info("TiDB cluster is synced",
					"extraGroup", extraGroup.Name,
					"tidbClusterName", obj.Name)
				return
			}
			logger.Info("TiDB cluster is not synced",
				"extraGroup", extraGroup.Name,
				"tidbClusterName", obj.Name,
				"diff", diff)
		}()

		obj.SetLabels(tc.GetLabels())
		for k, v := range tc.GetAnnotations() {
			obj.Annotations[k] = v
		}

		// if creation timestamp is zero, these code only works for new tc
		if obj.CreationTimestamp.IsZero() {
			logger.Info("creating new TiDB cluster for extra group",
				"extraGroup", extraGroup.Name,
				"tidbClusterName", obj.Name)

			obj.Spec = tc.Spec
			initializeGKEV1MinDomains(t.Cluster, obj)

			if tidbMountOOMPathForNew {
				logger.Info("mounting OOM path to host for new TiDB cluster",
					"extraGroup", extraGroup.Name)
				c.MountOOMPathToHost(obj)
			}
			if tidbMountReplayerPathForNew {
				logger.Info("mounting plan replayer path to host for new TiDB cluster",
					"extraGroup", extraGroup.Name)
				c.MountPlanReplayerPathToHost(obj)
			}

			// NOTE: the default TiDB Node Group (tidbcluster.go) sets PriorityClassName out of this creation statements.
			//   We add this here because we need to ensure the existing extra TiDB Node Groups wouldn't be set this to avoid pods restart.
			if t.Cluster.FeatureEnabled(consts.ClusterFeatureSetPriorityClass) {
				logger.Info("setting priority class for new TiDB cluster",
					"extraGroup", extraGroup.Name,
					"priorityClassName", consts.PriorityClasName)
				obj.Spec.PriorityClassName = ptr.To(consts.PriorityClasName)
			}

			// cluster has been initialized in the main cluster.
			logger.Info("applying statement log overlay for new TiDB cluster",
				"extraGroup", extraGroup.Name)
			statementlog.Overlay(obj)

			logger.Info("reconciling audit configuration for new TiDB cluster",
				"extraGroup", extraGroup.Name)
			externalID := t.GetAWSAuditLogRoleExternalID()
			if _, err = audit.Reconcile(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), extraGroup); err != nil {
				logger.Error(err, "failed to reconcile audit for new TiDB cluster",
					"extraGroup", extraGroup.Name)
				return err
			}

			// TODO(ideascf): it seems like that we don't need to return here but just fallthrough
			return nil
		}

		if t.Cluster.Status.Phase != "" && t.Cluster.Status.Phase != cloudv1.ClusterPhasePending &&
			t.Cluster.Status.Phase != cloudv1.ClusterPhaseCreating {
			// set MinReadySeconds after cluster created
			// NOTE: when the cluster becomes running, one more round of reconcile is needed to trigger this
			if err = c.OverlayMinReadyAnnotations(ctx, extraGroup.Name, obj); err != nil {
				return err
			}
		}

		// Set preStop hook for tidb when the tiproxy is disabled
		if t.Cluster.FeatureEnabled(consts.ClusterFeatureEnableGracefulShutdown) && extraGroup.Spec.TiProxy.Replicas == 0 {
			setPreStopHook(ctx, t.Cluster, extraGroup.Name, obj, t.Config.ClusterController)
		}

		// Set PV retain policy to 'Delete'
		pvPolicy := corev1.PersistentVolumeReclaimDelete
		obj.Spec.PVReclaimPolicy = &pvPolicy

		// set config update strategy
		obj.Spec.ConfigUpdateStrategy = tc.Spec.ConfigUpdateStrategy

		// FRM-2728: https://tidb.atlassian.net/browse/FRM-2728
		// Enable PPv2 For TiDB
		mutateTcForComponentsEnablePPv2(ctx, t.Cluster, obj)

		// ensure endpoint update take effect
		if obj.Spec.TiDB != nil && tc.Spec.TiDB != nil {
			obj.Spec.TiDB.Service = tc.Spec.TiDB.Service
		}

		// update replicas, storage-size for non instance-type scaling components
		comps := t.NonInstanceTypeScalingComponents(extraGroup)
		// it will carefully deal with scaling in by setting the delete slot to avoid az imbalance
		if err := t.mutateTcForReplicas(ctx, cli, obj, tc, comps); err != nil {
			return err
		}

		// do not update PD if PD is unavailable as it will call PD HTTP API,
		// and rely on requeue in PostCreate if cluster is not running (because PD is unavailable).
		if err = t.ConfigManager.UpdateConfig(ctx, t.Cluster, t.IsRegionalManaged(), configmanager.OverrideRevisionFinder(t.Local),
			obj, t.PDClient, t.removeUnavailablePD(comps)); err != nil {
			return err
		}

		// NOTE: we should call `mutateTiDBForTokenBasedAuth` after `t.ConfigManager.UpdateConfig` to ensure
		// the version of components in `obj` is CORRECT(patched/updated by ConfigRevision).
		mutateTiDBForTokenBasedAuth(ctx, t.Cluster, obj, tc)
		mutateTiDBAndTiProxyForTiProxy(t.Cluster, &extraGroup.Spec.TiProxy, obj, tc, extraGroup.Name)

		// mutate component if a rolling-update is needed after instance-type changed (and new nodegroups are ready)
		if err = t.mutateTcForInstanceType(ctx, cli, obj, tc, cli, extraGroup); err != nil {
			return err
		}
		t.mutateTcForSuspendActions(obj, tc)
		if err := overlayTiDBAddIndex(c, obj); err != nil {
			return err
		}

		if tidbMountOOMPathForExisting || tidbMountOOMPathForUpgrading {
			c.MountOOMPathToHost(obj)
		}
		if tidbMountReplayerPathForExisting || tidbMountReplayerPathForUpgrading {
			c.MountPlanReplayerPathToHost(obj)
		}

		// Only overlay node selector for GKE clusters
		if t.Config.Provider == consts.ProviderTypeGKE {
			nodepool.OverlayNodeSelectorWithK8sVersion(obj, t.GetNodeVersion(true))
		}

		// Merge the sidecar update and the tidb component update together
		externalID := t.GetAWSAuditLogRoleExternalID()
		if err := audit.FillSidecarContainer(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), extraGroup); err != nil {
			return err
		}

		return nil
	})

	defer func() {
		// any error occurred, it's better to observe the current status of tidbcluster
		err2 := t.observeExtraTidbCluster(ctx, cli, extraGroup)
		if err != nil {
			// log the observe result but do not overwrite the original one
			if err2 != nil {
				t.Logger.Info("observe tidb cluster when creating/updating tc failed", "observeResult", err2)
			}
			return
		}
	}()

	if err != nil {
		// skip operations (e.g `setPDLowSpaceRatio`) in `PostCreate`
		return err
	}

	if opResult != controllerutil.OperationResultCreated {
		// When tc not create, audit.Reconcile will just update tc as what it is expect.
		// When tc already created. Since `volumeClaimTemplates` of StatefulSet do not support updated.
		// To enable audit log, it will need some trivial hack which will update the STS directly.
		// See comment of `audit.Reconcile` for detail.

		externalID := t.GetAWSAuditLogRoleExternalID()
		res, err := audit.Reconcile(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), extraGroup)
		if err != nil {
			return err
		}
		if !res.IsZero() {
			return fmt.Errorf("retry audit reconcile for extragroup tidbcluster %s", extraGroup.Name)
		}
	}

	return nil
}

func (t *TiDBGroup) observeExtraTidbCluster(ctx context.Context, remoteCli client.Client, extraGroup *cloudv1.ExtraGroupCluster) error {
	logger := log.FromContext(ctx)

	extraGroupNamer := t.TidbGroupNamer(extraGroup.Spec).
		WithComponent(string(cloudv1.ComponentTiDB))

	tc := &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      extraGroupNamer.GetTiDBName(),
			Namespace: extraGroupNamer.GetTiDBNamespace(),
		},
	}

	logger.Info("observing extra TiDB cluster",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"namespace", tc.Namespace)

	key := types.NamespacedName{Name: extraGroupNamer.GetTiDBName(), Namespace: extraGroupNamer.GetTiDBNamespace()}
	if err := remoteCli.Get(ctx, key, tc); err != nil {
		logger.Error(err, "failed to get TiDB cluster",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return err
	}

	logger.Info("observing extra group cluster status",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"tidbClusterConditions", len(tc.Status.Conditions))

	t.ExtraGroupTCGetter[extraGroup.Name] = tidbcluster.NewV1Getter(tc)
	// Error from resolving the cluster status does not block the remaining tasks
	err := t.ObserveExtraGroupClusterStatus(ctx, tc, extraGroup)
	if err != nil {
		logger.Error(err, "failed to observe extra group cluster status",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return err
	}

	// To avoid the case that the extraGroup is not synced but the `isClusterReady` returns true,
	// check whether the extraGroup is synced before calling `isClusterReady`.
	if !t.Cluster.ExtraGroupSynced(extraGroup) {
		logger.Info("extra group is not synced yet",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return fmt.Errorf("extraGroup is not synced")
	}

	logger.Info("checking if extra TiDB cluster is ready",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name)

	ready, err := t.isClusterReady(ctx, extraGroup, tc)
	if err != nil {
		logger.Error(err, "failed to check cluster readiness",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return fmt.Errorf("isClusterReady(%s): %s", tc.Name, err)
	}
	if !ready {
		logger.Info("TiDB cluster is not ready yet",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return fmt.Errorf("tidb cluster(%s) is not ready yet", tc.Name)
	}

	logger.Info("extra TiDB cluster is ready",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name)

	return nil
}

func (t *TiDBGroup) ensureZeroReplicasExtraGroupCleaned(ctx context.Context, extraGroup *cloudv1.ExtraGroupCluster) error {
	remoteCli := t.EnvProvider.GetClient()

	if err := t.cleanupGKEHeterogeneousExtraTiDBGroupBeforeDelete(ctx, remoteCli, extraGroup); err != nil {
		return err
	}

	// remove extragroup status
	// if extragroup tidb replicas is 0, it means the extragroup is deleted, we can just reset the status of extragroup,
	// and make full use of getDeletedObject and DeleteObject to ensure the deletion of TiDBCluster.
	// if the deletion failed, it can be noticed in the cluster event log by GetDeletedObject function error message.
	if isExtraGroupStatusExsisted(extraGroup) {
		// it only reset fields managed by TiDBCluster task, other fields such as lb status should be managed in extragroup.service task
		resetExtraGroupStatusForTiDBCluster(t.Cluster, extraGroup)
	}
	extraGroupNamer := t.TidbGroupNamer(extraGroup.Spec).
		WithComponent(string(cloudv1.ComponentTiDB))

	tc := &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      extraGroupNamer.GetTiDBName(),
			Namespace: extraGroupNamer.GetTiDBNamespace(),
		},
	}
	if err := k8s.DeleteObjectCompleted(ctx, remoteCli, tc); err != nil {
		return err
	}

	return nil
}

func (t *TiDBGroup) cleanupGKEHeterogeneousExtraTiDBGroupBeforeDelete(ctx context.Context, remoteCli client.Client, extraGroup *cloudv1.ExtraGroupCluster) error {
	groupID := extraGroup.Name
	if groupID == "" || groupID == t.Cluster.GetDefaultGroupID() {
		return nil
	}
	if t.Cluster.GetProviderType() != cloudv1.ProviderTypeGKE || !t.Cluster.TiDBGKEComputeClassIntentConfigured(groupID) {
		return nil
	}

	if err := gkeheterogeneous.DeleteGatekeeperAssignForComponentIfExists(ctx, remoteCli, t.Cluster.Name, cloudv1.ComponentTiDB, groupID); err != nil {
		return errors.Wrapf(err, "delete GKE heterogeneous gatekeeper assign for TiDB group %s", groupID)
	}
	t.Cluster.MutateAnnotations(func(annotations map[string]string) {
		delete(annotations, t.Cluster.TiDBGKEComputeClassBindingsAnnoKey(groupID))
	})
	return nil
}

// Methods that implement the Cluster interface

func (t *TiDBGroup) buildTidbCluster(c *ClusterContext, extraGroup *cloudv1.ExtraGroupCluster) (*opv1alpha1.TidbCluster, error) {
	fakeCluster := withTiproxyOverlayedCluster(t.Cluster, extraGroup)
	cc, err := getClusterConfigIn3layers(t.Local, fakeCluster)
	if err != nil {
		return nil, errors.Annotate(err, "error get cluster config")
	}

	tc, err := buildBasicTidbCluster(t.Cluster, cc)
	if err != nil {
		return nil, err
	}
	if err := c.buildTiDBConfig(tc, cc); err != nil {
		return nil, err
	}
	syncFeatureAnnotations(t.Cluster, tc)
	if err := t.Overlay(extraGroup, cc, tc); err != nil {
		return nil, err
	}
	return tc, nil
}

// isClusterReady checks whether the tidb cluster is ready
// 1. check whether the tc is ready(If ready condition is true, it means all components are ready)
// 2. check whether the tidb pods are ready(If ready condition is true, it means readiness gates are ready and tidb service is ready to accept traffic)
func (t *TiDBGroup) isClusterReady(ctx context.Context, extraGroup *cloudv1.ExtraGroupCluster, tc *opv1alpha1.TidbCluster) (bool, error) {
	logger := log.FromContext(ctx)

	logger.Info("checking TiDB cluster readiness conditions",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"conditionCount", len(tc.Status.Conditions))

	var ready bool
	for _, c := range tc.Status.Conditions {
		if c.Type == opv1alpha1.TidbClusterReady && c.Status == corev1.ConditionTrue {
			ready = true
			break
		}
	}
	if !ready {
		logger.Info("TiDB cluster ready condition is not true",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name,
			"conditions", tc.Status.Conditions)
		return false, fmt.Errorf("tc is not ready yet")
	}

	logger.Info("TiDB cluster ready condition is true, checking pod readiness",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name)

	remoteCli := t.EnvProvider.GetClient()
	labelProvider := t.newTidbGroupLabelProvider(extraGroup)

	pods, err := k8s.SelectPods(ctx, remoteCli, tc.Namespace, labelProvider.GetComponentPodLabels(cloudv1.ComponentTiDB))
	if err != nil {
		logger.Error(err, "failed to select TiDB pods",
			"extraGroup", extraGroup.Name,
			"tidbClusterName", tc.Name)
		return false, fmt.Errorf("failed to select tidb pods: %v", err)
	}

	logger.Info("found TiDB pods for readiness check",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"podCount", len(pods),
		"memberCount", len(tc.Status.TiDB.Members))

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

	for name := range tc.Status.TiDB.Members {
		pod, ok := name2Pods[name]
		if !ok {
			logger.Info("TiDB pod not found",
				"extraGroup", extraGroup.Name,
				"tidbClusterName", tc.Name,
				"podName", name)
			return false, fmt.Errorf("tidb pod(%s) is not found", name)
		}
		if !k8s.IsPodReady2(pod) {
			logger.Info("TiDB pod is not ready",
				"extraGroup", extraGroup.Name,
				"tidbClusterName", tc.Name,
				"podName", name,
				"podPhase", pod.Status.Phase)
			return false, fmt.Errorf("tidb pod(%s) is not ready", name)
		}
	}

	logger.Info("all TiDB pods are ready",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", tc.Name,
		"readyPodCount", len(tc.Status.TiDB.Members))

	return true, nil
}

// TODO(ideascf): since the `getClusterConfigIn3layers` only returns configurations for components with replicas > 0,
// but the tiproxy may not be deployed in the default tidbgroup.
// withTiproxyOverlayedCluster set the tiproxy spec from the tidbgroup to a clone of this cluster.
// This hack is used for some underlying function depends on the Cluster and it's not easy to refactor.
// For example,
//  1. the `getClusterConfigIn3layers` only returns configurations for components with replicas > 0,
//     but the tiproxy may not be deployed in the default tidbgroup.
func withTiproxyOverlayedCluster(cluster *cloudv1.Cluster, extraGroup *cloudv1.ExtraGroupCluster) *cloudv1.Cluster {
	clonedCluster := cluster.DeepCopy()
	clonedCluster.Spec.Cluster.TiProxy = extraGroup.Spec.TiProxy
	return clonedCluster
}

var _ provider.Cluster = &tidbGroupLabelProvider{}

type tidbGroupLabelProvider struct {
	namer      *namer.Namer
	extraGroup *cloudv1.ExtraGroupCluster
}

func (t *tidbGroupLabelProvider) GetComponentNodeLabels(componentType cloudv1.ComponentType) map[string]string {
	// these labels will be used to map pod to zone by node name,
	// so no instance-type label added here so that we can get all nodes
	return t.namer.WithComponent(string(componentType)).Labels()
}

// GetComponentPodLabels implements the provider.Cluster interface.
func (t *tidbGroupLabelProvider) GetComponentPodLabels(componentType cloudv1.ComponentType) map[string]string {
	return t.namer.WithComponent(string(componentType)).GetPodLabels()
}

func (t *TiDBGroup) newTidbGroupLabelProvider(extraGroup *cloudv1.ExtraGroupCluster) provider.Cluster {
	return &tidbGroupLabelProvider{
		namer:      t.TidbGroupNamer(extraGroup.Spec),
		extraGroup: extraGroup,
	}
}

// removed functions:
// overlayTiFlashAndTiProxy
// overlayTiKVForDedicatedRaftEngineStorage
// overlayTiKVKmsCredentials

// Overlay implements the provider.Cluster interface.
func (t *TiDBGroup) Overlay(extraGroup *cloudv1.ExtraGroupCluster, clusterConfig *api.DescribeClusterConfigOutput, tc *opv1alpha1.TidbCluster) error {
	cctx, err := t.buildClusterContext(extraGroup)
	if err != nil {
		return err
	}

	fs := []func(t *TiDBGroup, tc *opv1alpha1.TidbCluster) error{
		func(t *TiDBGroup, tc *opv1alpha1.TidbCluster) error {
			return OverlayTiProxy(t.BaseTask, cctx, &extraGroup.Spec.TiProxy, tc, clusterConfig)
		},
		overlayTiDBGroupSuspendActions,
	}

	switch t.Config.Provider {
	case consts.ProviderTypeEKS:
		// do nothing
	case consts.ProviderTypeGKE:
		fs = append(fs, overlayTiDBGroupNodeSelectorWithInstanceType)
		if t.Cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) {
			fs = append(fs, overlayTiDBGroup3LayerArchToleration)
			fs = append(fs, overlayTiDBGroup3LayerArchNodeSelector)
		}
		fs = append(fs, overlayTiDBGroupArm64Toleration)
	case consts.ProviderTypeAKS:
		fs = append(fs, overlayTiDBGroupNodeSelectorWithInstanceType)
	}

	for _, f := range fs {
		if err := f(t, tc); err != nil {
			return err
		}
	}
	return nil
}

func (t *TiDBGroup) mutateTcForReplicas(ctx context.Context, cli client.Client, currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) error {
	logger := log.FromContext(ctx)

	logger.Info("mutating TiDB cluster for replicas",
		"tidbClusterName", currentTc.Name,
		"components", comps)

	err := t.preMutateTcReplicas(ctx, cli, currentTc, newTc, comps)
	if err != nil {
		logger.Error(err, "failed to pre-mutate TiDB cluster replicas",
			"tidbClusterName", currentTc.Name)
		return err
	}

	for _, comp := range comps {
		switch comp {
		case cloudv1.ComponentTiDB:
			currentReplicas := currentTc.Spec.TiDB.Replicas
			newReplicas := newTc.Spec.TiDB.Replicas

			logger.Info("updating TiDB replicas",
				"tidbClusterName", currentTc.Name,
				"currentReplicas", currentReplicas,
				"newReplicas", newReplicas)

			if !t.Cluster.FeatureEnabled(consts.ClusterFeatureNoScaleOutTiDBInParallel) {
				if newReplicas > currentReplicas {
					diff := newReplicas - currentReplicas
					currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism = &diff
					logger.Info("setting TiDB scale out parallelism",
						"tidbClusterName", currentTc.Name,
						"parallelism", diff)
				}
			} else if currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism != nil && *currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism > 1 {
				// Reset ScaleOutParallelism to the default value (1) if the feature gate is not enabled.
				logger.Info("resetting TiDB scale out parallelism to default",
					"tidbClusterName", currentTc.Name,
					"previousParallelism", *currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism)
				currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism = ptr.To(int32(1))
			}
			currentTc.Spec.TiDB.Replicas = newReplicas
			ensureGKEV1MinDomainsOnActivation(
				t.Cluster, currentTc, cloudv1.ComponentTiDB, currentReplicas, newReplicas,
			)

		case cloudv1.ComponentTiProxy:
			var currentReplicas, newReplicas int32
			if currentTc.Spec.TiProxy != nil {
				currentReplicas = currentTc.Spec.TiProxy.Replicas
			}
			if newTc.Spec.TiProxy != nil {
				newReplicas = newTc.Spec.TiProxy.Replicas
			}

			logger.Info("updating TiProxy replicas",
				"tidbClusterName", currentTc.Name,
				"currentReplicas", currentReplicas,
				"newReplicas", newReplicas)

			if currentTc.Spec.TiProxy == nil || currentTc.Spec.TiProxy.Replicas == 0 {
				if newTc.Spec.TiProxy != nil {
					logger.Info("scaling TiProxy from 0",
						"tidbClusterName", currentTc.Name,
						"newReplicas", newReplicas)
					currentTc.Spec.TiProxy = newTc.Spec.TiProxy // scaling from 0
				}
			}
			if currentTc.Spec.TiProxy != nil {
				if newTc.Spec.TiProxy == nil {
					logger.Info("scaling TiProxy to 0",
						"tidbClusterName", currentTc.Name)
					currentTc.Spec.TiProxy.Replicas = 0
				} else {
					currentTc.Spec.TiProxy.Replicas = newReplicas
				}
			}
			ensureGKEV1MinDomainsOnActivation(
				t.Cluster, currentTc, cloudv1.ComponentTiProxy, currentReplicas, newReplicas,
			)
		}
	}

	return nil
}

func (t *TiDBGroup) preMutateTcReplicas(ctx context.Context, cli client.Client, currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) error {
	for _, comp := range comps {
		var current, desired int32 = 0, 0
		switch comp {
		case cloudv1.ComponentTiDB:
			current = currentTc.Spec.TiDB.Replicas
			desired = newTc.Spec.TiDB.Replicas
		case cloudv1.ComponentTiProxy:
			if currentTc.Spec.TiProxy != nil {
				current = currentTc.Spec.TiProxy.Replicas
			}
			if newTc.Spec.TiProxy != nil {
				desired = newTc.Spec.TiProxy.Replicas
			}
		}
		err := scaleInComponentBalanced(ctx, cli, current, desired, currentTc, comp, t.Zones())
		if err != nil {
			return err
		}
	}
	return nil
}

func (t *TiDBGroup) mutateTcCompForInstanceType(
	ctx context.Context,
	cli client.Client,
	currentTc, newTc *opv1alpha1.TidbCluster,
	comps []cloudv1.ComponentType,
	remoteCli client.Client,
	extraGroup *cloudv1.ExtraGroupCluster,
) error {
	if err := t.mutateTcForReplicas(ctx, cli, currentTc, newTc, comps); err != nil {
		return err
	}
	var (
		nodeAffinityPolicy   = corev1.NodeInclusionPolicyIgnore
		topoSpreadConstraint = opv1alpha1.TopologySpreadConstraint{
			TopologyKey:        "topology.kubernetes.io/zone",
			MaxSkew:            1,
			NodeAffinityPolicy: &nodeAffinityPolicy,
		}
	)

	// set minDomains to zones count for azure cluster
	if t.Config.Provider == consts.ProviderTypeAKS {
		topoSpreadConstraint.MinDomains = ptr.To(int32(len(t.Zones())))
	}

	for _, comp := range comps {
		switch comp {
		case cloudv1.ComponentTiDB:
			currentTc.Spec.TiDB.ResourceRequirements = newTc.Spec.TiDB.ResourceRequirements // no storage
			// reset `GOMEMLIMIT` when changing resource limit, ref `AssembleTidbClusterValues`
			// NOTE: we do not append if `GOMEMLIMIT` is not found, as the situation is unknown.
			tidbResource := newTc.Spec.TiDB.ResourceRequirements
			if tidbResource.Limits.Memory() != nil && !tidbResource.Limits.Memory().IsZero() {
				tidbGoLimit := int64(GOMEMLIMIT_RATE * float64(tidbResource.Limits.Memory().Value()))
				for i, env := range currentTc.Spec.TiDB.Env {
					if env.Name == "GOMEMLIMIT" {
						currentTc.Spec.TiDB.Env[i] = corev1.EnvVar{Name: "GOMEMLIMIT", Value: strconv.FormatInt(tidbGoLimit, 10)}
						break
					}
				}
			}

			if t.useClusterAutoscaler() {
				mt := extraGroup.Spec.TiDB.InstanceType
				currentTc.Spec.TiDB.NodeSelector[consts.InstanceTypeLabelKey] = mt
			}
		case cloudv1.ComponentTiProxy:
			if newTc.Spec.TiProxy != nil {
				currentTc.Spec.TiProxy.NodeSelector = newTc.Spec.TiProxy.NodeSelector
				currentTc.Spec.TiProxy.Tolerations = newTc.Spec.TiProxy.Tolerations
				currentTc.Spec.TiProxy.StorageClassName = newTc.Spec.TiProxy.StorageClassName
				currentTc.Spec.TiProxy.Requests[corev1.ResourceCPU] = newTc.Spec.TiProxy.Requests[corev1.ResourceCPU]
				currentTc.Spec.TiProxy.Requests[corev1.ResourceMemory] = newTc.Spec.TiProxy.Requests[corev1.ResourceMemory]
				currentTc.Spec.TiProxy.Limits[corev1.ResourceMemory] = newTc.Spec.TiProxy.Limits[corev1.ResourceMemory]
				// NOTE: scale up TiProxy if needed

				if t.useClusterAutoscaler() {
					it := extraGroup.Spec.TiProxy.InstanceType
					currentTc.Spec.TiProxy.NodeSelector[consts.InstanceTypeLabelKey] = it
					currentTc.Spec.TiProxy.TopologySpreadConstraints = []opv1alpha1.TopologySpreadConstraint{topoSpreadConstraint}
				}
			}
		}
	}
	if err := applyInstanceTypeMinDomains(
		ctx, t.Cluster, remoteCli, currentTc, extraGroup.Name, comps...,
	); err != nil {
		return errors.Wrap(err, "prepare GKE v1 minDomains for instance-type change")
	}

	return t.ConfigManager.UpdateConfig(ctx, t.Cluster, t.IsRegionalManaged(), configmanager.OverrideRevisionFinder(t.Local),
		currentTc, t.PDClient, t.removeUnavailablePD(comps))
}

func (t *TiDBGroup) useClusterAutoscaler() bool {
	return t.Config.Provider == consts.ProviderTypeGKE || t.Config.Provider == consts.ProviderTypeAKS
}

func (t *TiDBGroup) mutateTcForInstanceType(ctx context.Context, cli client.Client, currentTc, newTc *opv1alpha1.TidbCluster, remoteCli client.Client, extraGroup *cloudv1.ExtraGroupCluster) error {
	logger := t.Logger.WithValues("group", extraGroup.Name)
	comps := t.Cluster.InstanceTypeChangedComponent(newTc.Name)
	if len(comps) == 0 {
		return nil
	}

	compsCloudV1 := make([]cloudv1.ComponentType, 0, len(comps))

	logger.Info("updating TiDB cluster config for instance type",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", newTc.Name,
		"components", comps)

	for _, comp := range comps {
		switch comp {
		case cloudv1.ComponentTiDB:
			compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiDB)
			if currentTc.Spec.TiDB.Annotations == nil {
				currentTc.Spec.TiDB.Annotations = make(map[string]string)
			}
			currentTc.Spec.TiDB.Annotations[consts.TcInstanceTypeAnnotationKey] = extraGroup.Spec.TiDB.InstanceType
			logger.Info("updating TiDB instance type annotation",
				"extraGroup", extraGroup.Name,
				"instanceType", extraGroup.Spec.TiDB.InstanceType)
		case cloudv1.ComponentTiProxy:
			if currentTc.Spec.TiProxy != nil {
				compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiProxy)
				if currentTc.Spec.TiProxy.Annotations == nil {
					currentTc.Spec.TiProxy.Annotations = make(map[string]string)
				}
				currentTc.Spec.TiProxy.Annotations[consts.TcInstanceTypeAnnotationKey] = extraGroup.Spec.TiProxy.InstanceType
				logger.Info("updating TiProxy instance type annotation",
					"extraGroup", extraGroup.Name,
					"instanceType", extraGroup.Spec.TiProxy.InstanceType)
			}
		}
	}

	logger.Info("mutating TiDB cluster components for instance type",
		"extraGroup", extraGroup.Name,
		"tidbClusterName", newTc.Name,
		"componentsToUpdate", compsCloudV1)

	// update cluster's replcias, storage-size, storage-class, requests/limits and config (and image)
	// in one operation when restarting pods to avoid evicting-leaders, restarting-pods, etc twice
	return t.mutateTcCompForInstanceType(ctx, cli, currentTc, newTc, compsCloudV1, remoteCli, extraGroup)
}

// useCMEKStorageClass will use CMEK storage class for TiDB cluster
func (t *TiDBGroup) useCMEKStorageClass(tc *opv1alpha1.TidbCluster) error {
	if tc.Spec.TiDB != nil && tc.Spec.TiDB.StorageClassName != nil {
		tc.Spec.TiDB.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.TiDB.StorageClassName))
	}
	if tc.Spec.TiProxy != nil && tc.Spec.TiProxy.StorageClassName != nil {
		tc.Spec.TiProxy.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.TiProxy.StorageClassName))
	}
	return nil
}

func (t *TiDBGroup) NonInstanceTypeScalingComponents(extraGroup *cloudv1.ExtraGroupCluster) []cloudv1.ComponentType {
	db := extraGroup.Status

	comps := make([]cloudv1.ComponentType, 0)

	tidbScaling := func() bool {
		if db.TiDB != nil {
			if db.TiDB.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone {
				return true
			}
			for _, member := range db.TiDB.Members {
				if member.InstanceType != "" && !t.Cluster.TiDBMemberInstanceTypeAllowedForGroup(member.InstanceType, extraGroup.Name, extraGroup.Spec.TiDB.InstanceType) {
					return true
				}
			}
		}
		return false
	}
	tiproxyScaling := func() bool {
		if db.TiProxy != nil {
			if db.TiProxy.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone {
				return true
			}
			for _, member := range db.TiProxy.Members {
				if member.InstanceType != "" && member.InstanceType != extraGroup.Spec.TiProxy.InstanceType {
					return true
				}
			}
		}
		return false
	}

	if !tidbScaling() {
		comps = append(comps, cloudv1.ComponentTiDB)
	}
	if !tiproxyScaling() {
		comps = append(comps, cloudv1.ComponentTiProxy)
	}

	return comps
}

// mutateTcForSuspendActions mutates the TidbCluster for suspend actions.
func (t *TiDBGroup) mutateTcForSuspendActions(currentTc, newTc *opv1alpha1.TidbCluster) {
	if currentTc.Spec.TiDB != nil && newTc.Spec.TiDB != nil {
		currentTc.Spec.TiDB.SuspendAction = newTc.Spec.TiDB.SuspendAction
	}
}

func (t *TiDBGroup) removeUnavailablePD(comps []cloudv1.ComponentType) []cloudv1.ComponentType {
	// In extragroup, it rely on the main cluster to manage PD.
	if t.Cluster.Status.DB == nil || t.Cluster.Status.DB.PD == nil ||
		t.Cluster.Status.DB.PD.Status == cloudv1.ComponentUnavailable {
		ret := make([]cloudv1.ComponentType, 0)
		for _, comp := range comps {
			if comp != cloudv1.ComponentPD {
				ret = append(ret, comp)
			}
		}
		return ret
	}

	return comps
}

func distTaskSupported(version string) bool {
	// https://docs.pingcap.com/tidb/stable/tidb-distributed-execution-framework#usage
	return semver.Compare(version, minDistTaskSupportedVersion) >= 0
}

func overlayTiDBGroupNodeSelectorWithInstanceType(t *TiDBGroup, tc *opv1alpha1.TidbCluster) error {
	disableNodePoolDataPlaneLabel := t.Cluster.FeatureEnabled(consts.ClusterFeatureDisableNodePoolDataPlaneLabel)

	if tc.Spec.TiDB != nil {
		// it's used in AssembleTidbClusterValues. instance type should be set in reconcile function
		// for GKE and AKS, we need to reset nodeSelector
		tc.Spec.TiDB.NodeSelector = map[string]string{}
		if !disableNodePoolDataPlaneLabel {
			tc.Spec.TiDB.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane
		}

		tc.Spec.TiDB.Affinity.NodeAffinity = &corev1.NodeAffinity{}
		tc.Spec.TiDB.Tolerations = append(
			tc.Spec.TiDB.Tolerations,
			utils.TolerationWithComponentShared()...,
		)
	}
	return nil
}

func overlayTiDBGroup3LayerArchToleration(_ *TiDBGroup, tc *opv1alpha1.TidbCluster) error {
	if tc.Spec.TiDB != nil {
		tc.Spec.TiDB.Tolerations = append(
			tc.Spec.TiDB.Tolerations,
			*utils.TolerationWith3LayerArch(),
		)
	}

	return nil
}

func overlayTiDBGroup3LayerArchNodeSelector(_ *TiDBGroup, tc *opv1alpha1.TidbCluster) error {
	if tc.Spec.TiDB != nil {
		tc.Spec.TiDB.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue
	}

	return nil
}

// overlayTiDBGroupArm64Toleration adds kubernetes.io/arch=arm64:NoSchedule toleration
// for GKE ARM instances (e.g. c4a-*) in TiDBGroup.
func overlayTiDBGroupArm64Toleration(t *TiDBGroup, tc *opv1alpha1.TidbCluster) error {
	if tc.Spec.TiDB == nil {
		return nil
	}
	it := t.Cluster.Spec.Cluster.TiDB.InstanceType
	if it == "" {
		return nil
	}
	instanceSpec, err := t.Provider.FindInstanceType(it)
	if err != nil {
		return nil
	}
	if instanceSpec.Arch == consts.ARM64ArchName {
		tc.Spec.TiDB.Tolerations = append(
			tc.Spec.TiDB.Tolerations,
			*utils.TolerationWithArch(consts.ARM64ArchName),
		)
	}
	return nil
}