package cdc

import (
	"context"
	"fmt"
	"path"

	"github.com/pelletier/go-toml"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	operatorconfig "github.com/pingcap/tidb-operator/pkg/apis/util/config"
	corev1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

	"github.com/tidbcloud/eks-provider/controllers/reconciler"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	gitopsv2 "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	gitopsutils "github.com/tidbcloud/infra-provider/pkg/gitops-sdk/utils"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/pingcap/label"
)

type NewArchTiCDCCluster struct {
	ticdcCluster
}

func BuildNewArchTiCDCCluster(cls ticdcCluster) *NewArchTiCDCCluster {
	return &NewArchTiCDCCluster{
		ticdcCluster: cls,
	}
}

func (ticdc *NewArchTiCDCCluster) Observe(ctx context.Context, obj client.Object) (reconciler.ExternalObservation, error) {
	return ticdc.ticdcCluster.Observe(ctx, obj)
}

func (ticdc *NewArchTiCDCCluster) Create(ctx context.Context, obj client.Object) error {
	cr, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return errNotTiCDC
	}

	if err := ticdc.ensureStorageClassExists(ctx); err != nil {
		return fmt.Errorf("create storage class error: %w", err)
	}
	nodeLabels, err := ticdc.prepareNode(ctx, cr)
	if err != nil {
		return err
	}
	result, err := ticdc.deployTiCDC(ctx, cr, nodeLabels)
	if err != nil {
		return err
	}
	if result != controllerutil.OperationResultNone {
		ticdc.event.Eventf(obj, corev1.EventTypeNormal, "CDCCreated", "Create cdc in data-plane successful")
		// add resource key annotation means create process success end
		if cr.Annotations == nil {
			cr.Annotations = make(map[string]string)
		}
		cr.Annotations[TiCDCResourceKeyLabel] = TiCDCResourceKey(cr.Spec.InstanceType, cr.Spec.Replicas)
		if err := ticdc.cli.Update(ctx, cr); err != nil {
			return err
		}
		cr.Status.Phase = cloudv1.TiCDCPhaseCreating
		return ticdc.cli.Status().Update(ctx, cr)
	}

	return fmt.Errorf("waiting ticdc create ready")
}

func (ticdc *NewArchTiCDCCluster) Update(ctx context.Context, obj client.Object) error {
	cr, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return errNotTiCDC
	}
	tc := new(v1alpha1.TidbCluster)
	if err := ticdc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(cr),
		Namespace: ticdc.destNamespace,
	}, tc); err != nil {
		return errors.Annotate(err, "failed get tc cr")
	}

	// try to set resource labels to gcp node pools, to propagate the labels to vm instances
	// it only affects billings, and doesn't affect ticdc business, so just records log if error occurred
	if err := ticdc.cdc.SetResourceLabels(ctx, cr.DeepCopy()); err != nil {
		ticdc.log.Error(err, "set resource labels")
	}

	if cr.Spec.Configuration != nil {
		updated, err := ticdc.updateConfig(ctx, tc, cr.Spec.Configuration)
		if err != nil {
			ticdc.log.Error(err, "update configuration error")
		} else if updated {
			ticdc.log.Info("update configuration successfully, wait next reconcile")
			return nil
		}
	}

	return ticdc.syncScaling(ctx, cr, tc)
}

func (ticdc *NewArchTiCDCCluster) updateConfig(
	ctx context.Context, tc *v1alpha1.TidbCluster, conf *cloudv1.DPConfiguration,
) (updated bool, err error) {
	if err := ticdc.setClusterIgnoreCDCConfig(ctx); err != nil {
		return false, errors.Annotate(err, "set cluster ignore ticdc config")
	}

	newTC := tc.DeepCopy()
	updated, err = ticdc.generateConfig(newTC, conf)
	if err != nil {
		return false, err
	}
	if !updated {
		// no update
		return
	}

	// use rolling update strategy to ensure that ticdc pods can be restarted after config is changed
	rollingUpdateStrategy := v1alpha1.ConfigUpdateStrategyRollingUpdate
	newTC.Spec.TiCDC.ConfigUpdateStrategy = &rollingUpdateStrategy
	patch := client.MergeFrom(tc.DeepCopy())
	if err := ticdc.destClusterCli.Patch(ctx, newTC, patch); err != nil {
		return false, errors.Annotate(err, "patch ticdc config")
	}
	return
}

func (ticdc *NewArchTiCDCCluster) setClusterIgnoreCDCConfig(ctx context.Context) error {
	if _, ok := ticdc.referenceOwner.Annotations[cloudv1.AnnoTiCDCIgnoreConfigUpdate]; ok {
		return nil
	}

	// add annotation to make cluster controller ignore cdc config update
	cluster := ticdc.referenceOwner.DeepCopy()
	cluster.Annotations[cloudv1.AnnoTiCDCIgnoreConfigUpdate] = "True"
	if err := ticdc.cli.Update(ctx, cluster); err != nil {
		// if the cluster is in regional api server, we don't need to update it
		if apierrors.IsNotFound(err) {
			return nil
		}

		return errors.Annotate(err, "update cluster annotation")
	}
	return nil
}

func (ticdc *NewArchTiCDCCluster) generateConfig(
	tc *v1alpha1.TidbCluster, conf *cloudv1.DPConfiguration,
) (updated bool, err error) {
	// update configuration
	predefinedConfig := make(map[string]interface{})
	if tc.Spec.TiCDC.Config != nil {
		// predefinedConfig is immutable, it will override the corresponding fields of desiredConfig
		predefinedConfig = ticdc.getPredefinedConfig(tc.Spec.TiCDC.Config.GenericConfig)
	}

	desiredConfig := make(map[string]interface{})
	if conf.Config != "" {
		if err := toml.Unmarshal([]byte(conf.Config), &desiredConfig); err != nil {
			return false, errors.Annotate(err, "failed to unmarshal cdc config")
		}
	}
	newCDCConfig := &v1alpha1.CDCConfigWraper{
		GenericConfig: operatorconfig.New(gitopsutils.MergeMaps(desiredConfig, predefinedConfig)),
	}
	if !apiequality.Semantic.DeepEqual(tc.Spec.TiCDC.Config, newCDCConfig) {
		tc.Spec.TiCDC.Config = newCDCConfig
		updated = true
	}

	// update image
	if tc.Spec.TiCDC.Version == nil || *tc.Spec.TiCDC.Version != conf.Tag {
		tc.Spec.TiCDC.Version = &conf.Tag
		updated = true
	}
	if updated {
		// currently, most of ticdc pods are using the image stared with "gcr.io/pingcap-public/dbaas",
		// and override will replace the gcr image to corresponding regional image, which can result in ticdc restarted.
		// we should execute replacing when config changed or version upgraded.
		tc.Spec.TiCDC.BaseImage = ticdc.cdc.OverrideImage(ticdc.referenceOwner, conf.BaseImage)
	}

	return updated, nil
}

func (ticdc *NewArchTiCDCCluster) getPredefinedConfig(config *operatorconfig.GenericConfig) map[string]interface{} {
	predefinedConfig := operatorconfig.New(make(map[string]interface{}))
	for _, key := range predefinedConfigKeys {
		v := config.Get(key)
		if v != nil {
			predefinedConfig.Set(key, v.Interface())
		}
	}
	return predefinedConfig.MP
}

func (ticdc *NewArchTiCDCCluster) Delete(ctx context.Context, obj client.Object) (reconciler.ExternalDeletion, error) {
	tc := new(v1alpha1.TidbCluster)
	cr, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return reconciler.ExternalDeletion{}, errNotTiCDC
	}
	err := ticdc.getTicdcCluster(ctx, tc, cr)
	// tidbcluster already deleted, we can think cdc also removed
	if apierrors.IsNotFound(err) {
		return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
	}
	if err != nil {
		return reconciler.ExternalDeletion{}, errors.Annotate(err, "get tc failed")
	}

	if isDRTiCDC(cr) {
		// if cdc deploy to downstream, tc name is db-cdc.
		// if db-cdc not deleted, db-discovery pod will remain
		if tc.Spec.TiCDC.Replicas != 0 {
			tc.Spec.TiCDC.Replicas = 0
			if err = ticdc.destClusterCli.Update(ctx, tc); err != nil && client.IgnoreNotFound(err) != nil {
				ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FaildUpdateTiCDC", "failed update ticdc cluster replicas to 0.")
				return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "update ticdc cluster error")
			}
		}
		err = ticdc.getTicdcCluster(ctx, tc, cr)
		// tidbcluster already deleted, we can think cdc also removed
		if apierrors.IsNotFound(err) {
			return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
		}

		err = ticdc.destClusterCli.Delete(ctx, tc)
		if client.IgnoreNotFound(err) != nil {
			ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDC", "failed delete cdc in data-plane cluster")
			return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc")
		}
		err = destroyCoredns(ctx, ticdc.destClusterCli, ticdc.coreDNSName)
		if err != nil {
			ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCoredns", "failed delete coredns in data-plane cluster")
			return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove coredns")
		}
	} else {
		_, err = controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, tc, func() error {
			tc.Spec.TiCDC = nil
			return nil
		})
		if client.IgnoreNotFound(err) != nil {
			ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDC", "failed delete cdc in data-plane cluster")
			return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc")
		}
		err = ticdc.deleteAccountCredentials(ctx, cr)
		if client.IgnoreNotFound(err) != nil {
			ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteAccountSecret", "failed delete account secret")
			return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "failed delete account secret")
		}
	}
	// TODO remove this step when operator supported
	err = ticdc.deleteTiCDCAsts(ctx, tc.Name, tc.Namespace)
	if client.IgnoreNotFound(err) != nil {
		ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDCAsts", "failed delete cdc asts in data-plane cluster")
		return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc asts")
	}

	err = ticdc.deleteTicdcPvc(ctx, tc.Namespace)
	if client.IgnoreNotFound(err) != nil {
		ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteTicdcPVC", "failed delete ticdc storage")
	}

	// Notice: currently, in 3-layer arch, cluster controller will create cert for ticdc if ticdc share same tc with tidb cluster
	//  if we delete cert after cdc deleted and recreate it when new cdc created, it may create wrong cert, such as
	// cert of tidb cluster is self-assigned, but cdc cert is issued by AWS since compat service lost some cluster CR info
	// err = ticdc.deleteTiCDCComponentCert(ctx, tc.Namespace, tc.Name)
	// if client.IgnoreNotFound(err) != nil {
	// 	ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDCComponentCert", "failed delete cdc component cert in data-plane cluster")
	// 	return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc component cert")
	// }

	err = ticdc.cdc.ReleaseNodes(ctx, cr, ticdc.nodeGroupNamePrefix, labels.Set{
		TiCDCCrIdLabel:    string(cr.UID),
		TiCDCClusterLabel: *cr.Spec.CDCClusterID,
		common.Component:  string(cloudv1.ComponentTiCDC),
	})
	if client.IgnoreNotFound(err) != nil {
		ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDCNodes", "failed delete cdc nodes")
		return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc node")
	}

	err = ticdc.ensureTiCDCPodClear(ctx, tc)
	if client.IgnoreNotFound(err) != nil {
		ticdc.event.Eventf(obj, corev1.EventTypeWarning, "FailedEnsureTiCDCPodClear", "failed ensure TiCDC pods clear")
		return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc pods")
	}

	ticdc.event.Eventf(obj, corev1.EventTypeNormal, "CDCDeleted", "Delete cdc successful")
	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

/*
case 1: replica: 1->2, scalingOutReplicas
case 2: replica: 2->1, scalingInReplicas
case 3: instance type: 4C8G -> 8C32G, 8C32G -> 4C8G, scalingInstanceType
case 4: replica: 1->2, instance type: 4C8G -> 8C32G, scalingInstanceType -> scalingOutReplicas
case 5: replica: 2->1, instance type: 8C32G -> 4C8G, scalingInReplicas -> scalingInstanceType
case 6: replica: 1->2, instance type: 8C32G -> 4C8G, scalingInstanceType -> scalingOutReplicas
case 7: replica: 2->1, instance type: 4C8G -> 8C32G, scalingInReplicas -> scalingInstanceType
*/
func (ticdc *NewArchTiCDCCluster) syncScaling(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster) error {
	key, exist := cr.Annotations[TiCDCResourceKeyLabel]
	if !exist {
		// old ticdc does not have the resource key, not support scaling
		return nil
	}
	if !ticdc.checkTiCDCCanScaling(ctx, cr) {
		ticdc.log.Info(fmt.Sprintf("current ticdc %s/%s does not support scaling", cr.Namespace, cr.Name))
		return nil
	}
	curInstanceType, curReplicas, err := TiCDCResourceValue(key)
	if err != nil {
		return errors.Annotate(err, "failed get current instance type and replicas")
	}

	if cr.Spec.InstanceType != curInstanceType {
		if cr.Spec.Replicas >= curReplicas {
			// scaling instance type first, then go to scaling out replicas
			if err = ticdc.scalingInstanceType(ctx, cr, tc, curInstanceType); err != nil {
				return err
			}
		} else {
			// scaling in replicas first, then go to scaling instance type
			if err = ticdc.scalingInReplicas(ctx, cr, tc); err != nil {
				return err
			}
		}
	} else if cr.Spec.Replicas > curReplicas {
		if err = ticdc.scalingOutReplicas(ctx, cr, tc); err != nil {
			return err
		}
	} else if cr.Spec.Replicas < curReplicas {
		if err = ticdc.scalingInReplicas(ctx, cr, tc); err != nil {
			return err
		}
	}
	if (cr.Spec.InstanceType != curInstanceType) || (cr.Spec.Replicas != curReplicas) {
		return fmt.Errorf("waiting ticdc scaling finish, current: %s:%d, desired: %s:%d", curInstanceType, curReplicas, cr.Spec.InstanceType, cr.Spec.Replicas)
	}
	return nil
}

func (ticdc *NewArchTiCDCCluster) checkTiCDCCanScaling(ctx context.Context, cr *cloudv1.TiCDC) bool {
	_, exist := cr.Annotations[TiCDCResourceKeyLabel]
	if !exist {
		// old ticdc does not have the resource key, not support scaling
		return false
	}
	if value, exist := cr.Annotations[TiCDCScalingDisableLabel]; exist && value == "true" {
		// disable scaling
		return false
	}

	// check ticdc is base on az nodegroup
	filter := &NgFilter{
		SpecifyLabels: map[string]string{
			TiCDCCrIdLabel:    string(cr.UID),
			TiCDCClusterLabel: *cr.Spec.CDCClusterID,
			common.Component:  string(cloudv1.ComponentTiCDC),
		},
	}
	topology, err := ticdc.describeNodegroupTopology(ctx, cr, filter)
	if err != nil {
		return false
	}
	if len(topology) < DefaultTiCDCZoneNum {
		return false
	}

	return true
}

// scalingInReplicas only scaling in replicas
func (ticdc *NewArchTiCDCCluster) scalingInReplicas(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster) error {
	// 1. modify tc replicas and wait pod scale in
	patch := client.MergeFrom(tc.DeepCopy())
	tc.Spec.TiCDC.Replicas = int32(cr.Spec.Replicas)
	err := ticdc.destClusterCli.Patch(ctx, tc, patch)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed update TiCDC %s replicas to %d", cr.Name, tc.Spec.TiCDC.Replicas))
		return errors.Annotate(err, "failed scale TiCDC replicas")
	}
	if tc.Status.TiCDC.StatefulSet.ReadyReplicas != int32(cr.Spec.Replicas) {
		ticdc.log.Info(fmt.Sprintf("ticdc %s pod replicas scaling in, wait ticdc pod replicas", cr.Name))
		return fmt.Errorf("ticdc pod replicas scaling in, wait ticdc pod replicas")
	}

	// 2. get ticdc pod topology
	topology, err := ticdc.getTiCDCPodTopology(ctx, cr, tc)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s pod topology", cr.Name))
		return errors.Annotate(err, "failed get ticdc pod topology")
	}

	// 3. sync node topology to nodegroup
	err = ticdc.cdc.UpdateNodegroupTopology(ctx, cr, topology)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed update ticdc %s nodegroup", cr.Name))
		return errors.Annotate(err, "failed update ticdc nodegroup")
	}

	if !ticdc.ensureTiCDCReplicasReady(tc) {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s replicas ready", cr.Name))
		return fmt.Errorf("wait ticdc spec replicas equals ready replicas")
	}

	err = ticdc.cdc.EnsureAllNodegroupsReady(ctx, cr)
	if err != nil {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s nodegroup all ready", cr.Name))
		return errors.Annotate(err, "failed ensure ticdc nodegroup all ready")
	}

	// 4. update resource key annotation
	instanceType, _, err := TiCDCResourceValue(cr.Annotations[TiCDCResourceKeyLabel])
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s resource value", cr.Name))
		return errors.Annotate(err, "failed get ticdc resource value")
	}
	ticdc.event.Eventf(cr, corev1.EventTypeNormal, "ScalingInReplicas", "ticdc scale in replicas success")
	cr.Annotations[TiCDCResourceKeyLabel] = TiCDCResourceKey(instanceType, cr.Spec.Replicas)
	return ticdc.cli.Update(ctx, cr)
}

// scalingOutReplicas only scaling out replicas
func (ticdc *NewArchTiCDCCluster) scalingOutReplicas(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster) error {
	// 1. get zone topology of ticdc cluster
	topology, err := ticdc.getTiCDCPodTopology(ctx, cr, tc)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s pod topology", cr.Name))
		return errors.Annotate(err, "failed get ticdc pod topology")
	}

	// 2. calculate desired topology and scale out az node groups
	desired := calculateDesiredTopology(topology, cr.Spec.Replicas)
	ticdc.log.Info(fmt.Sprintf("begin update ticdc %s node topology %+v", cr.Name, desired))
	err = ticdc.cdc.UpdateNodegroupTopology(ctx, cr, desired)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed sync ticdc %s nodegroup", cr.Name))
		return errors.Annotate(err, "failed sync ticdc nodegroup")
	}

	// 3. modify tc replicas and wait pod scale success, ensure az balance
	// https://docs.pingcap.com/zh/tidb-in-kubernetes/dev/tidb-scheduler
	err = ticdc.scalingOutOne(ctx, cr, tc)
	if err != nil {
		ticdc.log.Info(fmt.Sprintf("waiting ticdc %s replcias scaling out to desired", cr.Name))
		return errors.Annotate(err, "waiting ticdc replcias scaling out")
	}

	// 4. recycle redundant nodes
	newTopo, err := ticdc.getTiCDCPodTopology(ctx, cr, tc)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s pod topology after scaling out", cr.Name))
		return errors.Annotate(err, "failed get ticdc pod topology after scaling out")
	}
	err = ticdc.cdc.UpdateNodegroupTopology(ctx, cr, newTopo)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed sync ticdc %s nodegroup for recycle", cr.Name))
		return errors.Annotate(err, "failed sync ticdc nodegroup for recycle")
	}
	if !ticdc.ensureTiCDCReplicasReady(tc) {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s replicas ready", cr.Name))
		return fmt.Errorf("wait ticdc spec replicas equals ready replicas")
	}
	err = ticdc.cdc.EnsureAllNodegroupsReady(ctx, cr)
	if err != nil {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s nodegroup all ready", cr.Name))
		return errors.Annotate(err, "failed ensure ticdc nodegroup all ready")
	}

	// 5. update resource key annotation
	instanceType, _, err := TiCDCResourceValue(cr.Annotations[TiCDCResourceKeyLabel])
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s resource value", cr.Name))
		return errors.Annotate(err, "failed get cr resource value")
	}
	ticdc.event.Eventf(cr, corev1.EventTypeNormal, "ScalingOutReplicas", "ticdc scale out replicas success")
	cr.Annotations[TiCDCResourceKeyLabel] = TiCDCResourceKey(instanceType, cr.Spec.Replicas)
	return ticdc.cli.Update(ctx, cr)
}

// scalingOutOne scale out one replica
func (ticdc *NewArchTiCDCCluster) scalingOutOne(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster) error {
	// 1. wait all ticdc pods ready
	if !ticdc.ensureTiCDCReplicasReady(tc) {
		return fmt.Errorf("wait ticdc spec replicas equals ready replicas")
	}

	// 2. update ticdc replica + 1
	if tc.Spec.TiCDC.Replicas < int32(cr.Spec.Replicas) {
		patch := client.MergeFrom(tc.DeepCopy())
		tc.Spec.TiCDC.Replicas += 1
		if err := ticdc.destClusterCli.Patch(ctx, tc, patch); err != nil {
			return errors.Annotate(err, "failed patch ticdc replicas")
		}
		ticdc.log.Info(fmt.Sprintf("success scale ticdc replica to %d", tc.Spec.TiCDC.Replicas))
		return fmt.Errorf("wait ticdc replicas scaling out")
	}
	return nil
}

// scalingInstanceType only scaling instance type
func (ticdc *NewArchTiCDCCluster) scalingInstanceType(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster, currentInstanceType string) error {
	// 1. get zone topology of ticdc cluster
	topology, err := ticdc.getTiCDCPodTopology(ctx, cr, tc)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s pod topology", cr.Name))
		return errors.Annotate(err, "failed get ticdc pod topology")
	}

	// 2. build desired az nodegroups base on current topology
	nodeLabels, err := ticdc.createNodegroup(ctx, cr, topology)
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed create new nodegroup for ticdc %s", cr.Name))
		return errors.Annotate(err, "failed create new nodegroup")
	}

	err = ticdc.cdc.EnsureAllNodegroupsReady(ctx, cr)
	if err != nil {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s nodegroup all ready", cr.Name))
		return errors.Annotate(err, "failed ensure ticdc nodegroup all ready")
	}

	// 3. modify tc NodeSelector label/resource limit and wait pod migrate success
	// same replicas migrate between two nodegroups, no need deploy one by one
	_, err = controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, tc, func() error {
		ticdcResource, err := gitops.NewHardcode().GetResourceSpec(ticdc.provider, cr)
		if err != nil {
			ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s resource limit", cr.Name))
			return errors.Annotate(err, "failed get ticdc resources limit")
		}
		tc.Spec.TiCDC.NodeSelector = nodeLabels
		tc.Spec.TiCDC.ResourceRequirements = ticdcResource.Resources
		for i := range tc.Spec.TiCDC.StorageVolumes {
			if tc.Spec.TiCDC.StorageVolumes[i].Name == TiCDCDefaultDirName {
				tc.Spec.TiCDC.StorageVolumes[i].StorageSize = fmt.Sprintf(
					"%dGi", int(cr.Spec.StorageSize.Value())/1024/1024/1024)
			}
		}
		if tc.Spec.TiCDC.Replicas > 1 && tc.Spec.TiCDC.Replicas%2 > 0 {
			// Now, the default maxSkew of tc is 1. If the ticdc pods are odd number, it may break the maxSkew.
			// For example, there are 3 ticdc pods, and their corresponding zones are a, b, b.
			// When the ticdc pods are rolling restart, they start at ticdc-2.
			// When the ticdc-1 is restarting, since its volume is in zone-b, it can only be scheduled to zone-b.
			// But both ticdc-2 and ticdc-1 all in zone-b breaks the maxSkew,
			// so the ticdc-1 will always be blocking in pending.
			// In fact, ticdc doesn't depend on topologySpreadConstraints, because we already ensure that
			// the skew between zones is not more than 1 in node provisioning step.
			// So we can set maxSkew to 2 to solve the problem.
			tc.Spec.TiCDC.TopologySpreadConstraints = []v1alpha1.TopologySpreadConstraint{
				{
					MaxSkew:     2,
					TopologyKey: k8s.ZoneLabel,
				},
			}
		}

		return nil
	})
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed update ticdc %s resource limit", cr.Name))
		return errors.Annotate(err, "failed update ticdc resource limit")
	}

	// 4. ensure current pod migrate to new nodegroup
	// all pod running does not mean all pod migrate success, maybe is an intermediate state, must check pod node selector and replicas all synced
	if err = ticdc.ensureTiCDCPodWithTheNodeSelector(ctx, tc, map[string]string{
		TiCDCResourceKeyLabel: TiCDCResourceKey(cr.Spec.InstanceType, cr.Spec.Replicas),
	}); err != nil {
		ticdc.log.Info(fmt.Sprintf("failed ensure ticdc %s pod node selector", cr.Name))
		return errors.Annotate(err, "failed ensure ticdc pod node selector")
	}
	if !ticdc.ensureTiCDCReplicasReady(tc) {
		ticdc.log.Info(fmt.Sprintf("waiting ticdc %s pod migrate new nodegroup", cr.Name))
		return errors.New("waiting ticdc pod migrate new nodegroup")
	}

	// 5. delete old nodegroup
	if err = ticdc.ensureNodegroupClear(ctx, cr, &NgFilter{
		SpecifyLabels: map[string]string{
			TiCDCCrIdLabel:    string(cr.UID),
			TiCDCClusterLabel: *cr.Spec.CDCClusterID,
			common.Component:  string(cloudv1.ComponentTiCDC),
		},
		ExcludeLabels: map[string]string{
			TiCDCResourceKeyLabel: TiCDCResourceKey(cr.Spec.InstanceType, cr.Spec.Replicas),
		},
	}); err != nil {
		ticdc.log.Info(fmt.Sprintf("waiting ticdc %s old nodegroup clear", cr.Name))
		return err
	}

	// 6. update resource key annotation
	_, replicas, err := TiCDCResourceValue(cr.Annotations[TiCDCResourceKeyLabel])
	if err != nil {
		ticdc.log.Error(err, fmt.Sprintf("failed get ticdc %s resource value", cr.Name))
		return errors.Annotate(err, "failed get ticdc resource value")
	}
	ticdc.event.Eventf(cr, corev1.EventTypeNormal, "ScalingInstanceType", "ticdc scale instance type success")
	cr.Annotations[TiCDCResourceKeyLabel] = TiCDCResourceKey(cr.Spec.InstanceType, replicas)
	return ticdc.cli.Update(ctx, cr)
}

func (ticdc *NewArchTiCDCCluster) deployTiCDC(ctx context.Context, cr *cloudv1.TiCDC, labels map[string]string) (controllerutil.OperationResult, error) {
	if isDRTiCDC(cr) {
		return ticdc.deployDRTiCDC(ctx, cr, labels)
	}
	return ticdc.deployNormalTiCDC(ctx, cr, labels)
}

func (ticdc *NewArchTiCDCCluster) deployNormalTiCDC(ctx context.Context, cr *cloudv1.TiCDC, labels map[string]string) (controllerutil.OperationResult, error) {
	// 1.make namespace if necessary
	if _, err := controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{Name: ticdc.destNamespace},
	}, func() error {
		return nil
	}); err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "get namespace failed")
	}

	// 2.get tidb cluster cr
	tc := new(v1alpha1.TidbCluster)
	if err := ticdc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(cr),
		Namespace: ticdc.destNamespace,
	}, tc); err != nil {
		if apierrors.IsNotFound(err) {
			return controllerutil.OperationResultNone, errors.Annotate(err, "get tc failed")
		}
	}

	// 3.create ticdc cert
	if result, err := ticdc.createTiCDCComponentCert(ctx, tc.Namespace, tc.Name, "", utils.ClusterPrefix); err != nil {
		ticdc.event.Eventf(cr, corev1.EventTypeWarning, "FailedCreateTiCDCComponentCert", "create cdc tls cert failed")
		return controllerutil.OperationResultNone, errors.Annotate(err, "create cdc tls cert failed")
	} else if result == controllerutil.OperationResultCreated {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "CreateTiCDCComponentCert", "Create cdc tls cert successful")
	}

	// 4.create or update tidb cluster cr to deploy ticdc cluster
	if tc.Spec.TiCDC == nil {
		_, err := controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, tc, func() error {
			if updateErr := ticdc.overlayTicdc(tc, cr, labels, ticdc.buildTiCDCPodCredentials(cr), &gitopsv2.Gitops{Cli: ticdc.cli}); updateErr != nil {
				return errors.Annotate(updateErr, "overlay ticdc to tc failed")
			}
			return nil
		})
		if err != nil {
			return controllerutil.OperationResultNone, errors.Annotate(err, "create or update tc with ticdc failed")
		}
		// need wait next reconcile for ticdc status ready replicas update
		return controllerutil.OperationResultNone, errors.Annotate(err, "wait ticdc status update")
	}

	// 6. deploy replicas one by one
	if err := ticdc.scalingOutOne(ctx, cr, tc); err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "wait ticdc deploy replicas")
	}

	return controllerutil.OperationResultCreated, nil
}

func (ticdc *NewArchTiCDCCluster) deployDRTiCDC(ctx context.Context, cr *cloudv1.TiCDC, labels map[string]string) (controllerutil.OperationResult, error) {
	// 1.ensure cross region dns
	corednsClusterIP, err := ticdc.deployCrossRegionDNS(ctx, cr)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "create and get coredns cluster svc ip")
	}

	// 2.make namespace if necessary
	if _, err := controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{Name: ticdc.destNamespace},
	}, func() error {
		return nil
	}); err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "get namespace failed")
	}

	// 3.build or get tidbcluster cr of dr ticdc
	tc := new(v1alpha1.TidbCluster)
	if err := ticdc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(cr),
		Namespace: ticdc.destNamespace,
	}, tc); err != nil {
		if apierrors.IsNotFound(err) {
			tc = ticdc.buildDRTidbClusterCr(cr, corednsClusterIP)
		} else {
			return controllerutil.OperationResultNone, errors.Annotate(err, "get tc failed")
		}
	}

	// 4.create cert for ticdc sink data
	upstreamNamespace := "tidb" + cr.Spec.UpstreamClusterName
	if result, err := ticdc.createUpstreamClusterClientCert(ctx, tc.Namespace, tc.Name, upstreamNamespace); err != nil {
		ticdc.event.Eventf(cr, corev1.EventTypeWarning, "FailedUpstreamClusterClientCert", "create cdc tls cert failed")
		return controllerutil.OperationResultNone, errors.Annotate(err, "create cdc tls cert failed")
	} else if result == controllerutil.OperationResultCreated {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "CreateUpstreamClusterClientCert", "Create upstream cluster client cert successful")
	}

	if result, err := ticdc.createCertForDiscovery(ctx, tc.Namespace, tc.Name, upstreamNamespace); err != nil {
		ticdc.event.Eventf(cr, corev1.EventTypeWarning, "FailedDiscoveryCert", "create discovery tls cert failed")
		return controllerutil.OperationResultNone, errors.Annotate(err, "create discovery tls cert failed")
	} else if result == controllerutil.OperationResultCreated {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "CreateDiscoveryCert", "Create discovery tls cert successful")
	}

	if result, err := ticdc.createTiCDCComponentCert(ctx, tc.Namespace, tc.Name, upstreamNamespace, utils.ClusterPrefix); err != nil {
		ticdc.event.Eventf(cr, corev1.EventTypeWarning, "FailedCreateTiCDCComponentCert", "create cdc tls cert failed")
		return controllerutil.OperationResultNone, errors.Annotate(err, "create cdc tls cert failed")
	} else if result == controllerutil.OperationResultCreated {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "CreateTiCDCComponentCert", "Create cdc tls cert successful")
	}

	// 5.create or update tidb cluster cr to deploy ticdc cluster
	if tc.Spec.TiCDC == nil {
		_, err = controllerutil.CreateOrUpdate(ctx, ticdc.destClusterCli, tc, func() error {
			if updateErr := ticdc.overlayTicdc(tc, cr, labels, nil, &gitopsv2.Gitops{Cli: ticdc.cli}); updateErr != nil {
				return errors.Annotate(updateErr, "overlay ticdc to tc failed")
			}
			return nil
		})
		if err != nil {
			return controllerutil.OperationResultNone, errors.Annotate(err, "create or update tc with ticdc failed")
		}
		// need wait next reconcile for ticdc status ready replicas update
		return controllerutil.OperationResultNone, errors.Annotate(err, "wait ticdc status update")
	}

	// 6. deploy replicas one by one
	if err = ticdc.scalingOutOne(ctx, cr, tc); err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "wait ticdc deploy replicas")
	}

	return controllerutil.OperationResultCreated, nil
}

func (ticdc *NewArchTiCDCCluster) deployCrossRegionDNS(ctx context.Context, cr *cloudv1.TiCDC) (string, error) {
	if cr.Spec.UpstreamCoreDNSIp == "" {
		return "", errors.New("empty core dns ips in ticdc cr")
	}
	coreDNSIps := cr.Spec.UpstreamCoreDNSIp

	if consts.ProviderTypeGKE == ticdc.providerType {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "deployKubeDNS", fmt.Sprintf("begin deploy kube-dns, %s", ticdc.providerType))
		if err := deployKubeDNS(ctx, ticdc.destClusterCli, ticdc.coreDNSName, cr.Spec.UpstreamClusterName, coreDNSIps); err != nil {
			return "", errors.Annotate(err, "deploy kubedns")
		}
	} else {
		ticdc.event.Eventf(cr, corev1.EventTypeNormal, "deployCoredns", fmt.Sprintf("begin deploy core-dns, %s", ticdc.providerType))
		if err := deployCoredns(ctx, ticdc.destClusterCli, ticdc.coreDNSName, cr.Spec.UpstreamClusterName, coreDNSIps, ticdc.providerType); err != nil {
			return "", errors.Annotate(err, "deploy coredns")
		}
	}

	corednsClusterIP, err := ensureCoreDNSReady(ctx, ticdc.destClusterCli, ticdc.coreDNSName)
	if err != nil {
		return "", errors.Annotate(err, "get coredns cluster svc ip")
	}
	return corednsClusterIP, nil
}

func (ticdc *NewArchTiCDCCluster) buildDRTidbClusterCr(cr *cloudv1.TiCDC, coreDNSClusterIp string) *v1alpha1.TidbCluster {
	pvDeletePolicy := corev1.PersistentVolumeReclaimDelete
	return &v1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getTiDBClusterNameForTiCDC(cr),
			Namespace: ticdc.destNamespace,
			// node scaler require the labels to query or build TiCDC object
			Labels: map[string]string{
				consts.ClusterTicdcLabelKey:        cr.Name,
				consts.ClusterNamespaceLabelKey:    cr.Namespace,
				consts.ClusterNameLabelKey:         ticdc.crossRegionClusterNameLabelValue,
				consts.ClusterNetworkLabelKey:      ticdc.destNetwork.Name,
				consts.UpstreamClusterNameLabelKey: cr.Spec.UpstreamClusterName,
			},
		},
		Spec: v1alpha1.TidbClusterSpec{
			PVReclaimPolicy: &pvDeletePolicy,
			// tidb-operator does not rolling update by default, it must be specified explicitly
			ConfigUpdateStrategy: v1alpha1.ConfigUpdateStrategyRollingUpdate,
			DNSConfig: &corev1.PodDNSConfig{
				Nameservers: []string{
					coreDNSClusterIp,
				},
				// cdc will lookup db-discovery endpoint
				Searches: []string{
					"svc.cluster.local",
					"cluster.local",
					"ec2.internal",
				},
				Options: []corev1.PodDNSConfigOption{
					{
						Name:  "ndots",
						Value: pointer.String("5"),
					},
				},
			},
			DNSPolicy: corev1.DNSNone,
			AcrossK8s: true,
			TLSCluster: &v1alpha1.TLSCluster{
				Enabled: true,
			},
			Cluster: &v1alpha1.TidbClusterRef{
				Namespace: "tidb" + cr.Spec.UpstreamClusterName,
				Name:      utils.ClusterPrefix,
			},
		},
	}
}

func (ticdc *NewArchTiCDCCluster) deleteAccountCredentials(ctx context.Context, cr *cloudv1.TiCDC) error {
	if consts.ProviderTypeGKE == ticdc.providerType {
		// current only use in sink gcs case
		secret := &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      TiCDCGcpCredentialsName(cr),
				Namespace: ticdc.destNamespace,
			},
		}
		err := ticdc.destClusterCli.Delete(ctx, secret)
		return client.IgnoreNotFound(err)
	}

	return nil
}

func (ticdc *NewArchTiCDCCluster) buildTiCDCPodCredentials(cr *cloudv1.TiCDC) *AccountCredentialData {
	if consts.ProviderTypeGKE == ticdc.providerType {
		return &AccountCredentialData{
			volume: corev1.Volume{
				Name: "gcp-credentials",
				VolumeSource: corev1.VolumeSource{
					Secret: &corev1.SecretVolumeSource{
						// br secret is the network gcp service account key in old arch and is the cluster
						// gcp service account key in new arch.
						// in new arch, we can't access the secret with gcp service account in infra-api-server anymore.
						// since cluster controller already copies this secret to data plane both in old arch and new arch,
						// we can use it in data plane directly.
						// TODO: use ticdc dedicated service account to replace it.
						SecretName: consts.BRGCPSecretName,
						Items: []corev1.KeyToPath{
							{Key: consts.BRGCPSecretDataKey, Path: GcpCredentialsPath},
						},
					},
				},
			},
			mount: corev1.VolumeMount{
				Name:      "gcp-credentials",
				MountPath: GcpCredentialsBase,
			},
			env: corev1.EnvVar{
				Name:  "GOOGLE_APPLICATION_CREDENTIALS",
				Value: path.Join(GcpCredentialsBase, GcpCredentialsPath),
			},
		}
	}

	return nil
}

func (ticdc *NewArchTiCDCCluster) ensureStorageClassExists(ctx context.Context) error {
	sc := ticdc.cdc.BuildStorageClass()
	if err := ticdc.destClusterCli.Create(ctx, sc); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return nil
		}
		return err
	}
	return nil
}

func (ticdc *NewArchTiCDCCluster) prepareNode(ctx context.Context, cr *cloudv1.TiCDC) (map[string]string, error) {
	zones, err := ticdc.provider.FindRegionZones(ticdc.destNetwork.Spec.Region)
	if err != nil {
		return nil, err
	}
	if len(zones) < DefaultTiCDCZoneNum {
		return nil, fmt.Errorf("network support zones less than %d", DefaultTiCDCZoneNum)
	}

	topology, err := buildTopology(zones[:DefaultTiCDCZoneNum], cr.Spec.Replicas)
	if err != nil {
		return nil, err
	}
	return ticdc.createNodegroup(ctx, cr, topology)
}

func (ticdc *NewArchTiCDCCluster) createNodegroup(ctx context.Context, cr *cloudv1.TiCDC, topology map[string]int) (ngLabels map[string]string, err error) {
	for zone, nodeNum := range topology {
		name := TiCDCNodeGroupName(cr, zone)
		set := &nodegroup.Set{
			Name: name,
			NodeGroups: []cloudv1.NodegroupSpec{
				{
					Name: name,
					Labels: map[string]string{
						TiCDCCrIdLabel:        string(cr.UID),
						TiCDCClusterLabel:     *cr.Spec.CDCClusterID,
						TiCDCZoneLabel:        zone,
						TiCDCResourceKeyLabel: TiCDCResourceKey(cr.Spec.InstanceType, cr.Spec.Replicas),
					},
					MachineSpec: &cloudv1.MachineSpec{
						MachineType: cr.Spec.InstanceType,
					},
					Taints:   utils.Taint(cloudv1.ComponentTiCDC),
					Maximum:  int32(nodeNum),
					Minimum:  int32(nodeNum),
					Zones:    []string{zone},
					NodeRole: cr.Spec.NodeRole,
				},
			},
		}
		ngLabels, err = ticdc.cdc.CreateCdcNodeGroup(ctx, cr, ticdc.destCluster, set)
		if err != nil {
			return nil, err
		}
		// do not specify these label, it will cause schedule fail
		delete(ngLabels, TiCDCNodegroupNameLabel)
		delete(ngLabels, TiCDCZoneLabel)
	}

	return ngLabels, nil
}

func (ticdc *NewArchTiCDCCluster) describeNodegroupTopology(ctx context.Context, cr *cloudv1.TiCDC, filter *NgFilter) (map[string]int, error) {
	selector := labels.NewSelector()
	for k, v := range filter.SpecifyLabels {
		specify, err := labels.NewRequirement(k, selection.Equals, []string{v})
		if err != nil {
			return nil, err
		}
		selector = selector.Add(*specify)
	}
	return ticdc.cdc.GetNodegroupsByFilter(ctx, cr, selector)
}

func (ticdc *NewArchTiCDCCluster) ensureNodegroupClear(ctx context.Context, cr *cloudv1.TiCDC, filter *NgFilter) error {
	selector := labels.NewSelector()
	for k, v := range filter.SpecifyLabels {
		specify, err := labels.NewRequirement(k, selection.Equals, []string{v})
		if err != nil {
			return err
		}
		selector = selector.Add(*specify)
	}
	for k, v := range filter.ExcludeLabels {
		exclude, err := labels.NewRequirement(k, selection.NotEquals, []string{v})
		if err != nil {
			return err
		}
		selector = selector.Add(*exclude)
	}
	ticdc.log.Info(selector.String())
	return ticdc.cdc.DeleteNodegroupsByFilter(ctx, cr, selector)
}

func (ticdc *NewArchTiCDCCluster) ensureTiCDCPodWithTheNodeSelector(ctx context.Context, tc *v1alpha1.TidbCluster, nodeSelector map[string]string) error {
	podList := &corev1.PodList{}
	err := ticdc.destClusterCli.List(ctx, podList, client.InNamespace(tc.Namespace), client.MatchingLabels(map[string]string{
		label.ComponentLabelKey: string(cloudv1.ComponentTiCDC),
		label.InstanceLabelKey:  tc.Name,
	}))
	if client.IgnoreNotFound(err) != nil {
		return err
	}

	if tc.Spec.TiCDC.Replicas != int32(len(podList.Items)) {
		return fmt.Errorf("pod num %d dose not equals to tc replicas %d", len(podList.Items), tc.Spec.TiCDC.Replicas)
	}

	if nodeSelector != nil {
		for _, pod := range podList.Items {
			for k, v := range nodeSelector {
				value, exist := pod.Spec.NodeSelector[k]
				if !exist {
					return fmt.Errorf("ticdc pod %s node selector dose not contain key %s", pod.Name, k)
				}
				if v != value {
					return fmt.Errorf("ticdc pod %s node selector %s:%s not equals to desired %s:%s", pod.Name, k, value, k, v)
				}
			}
		}
	}

	return nil
}

// get current ticdc pod topology map[zone]nodeNum
func (ticdc *NewArchTiCDCCluster) getTiCDCPodTopology(ctx context.Context, cr *cloudv1.TiCDC, tc *v1alpha1.TidbCluster) (map[string]int, error) {
	podList := &corev1.PodList{}
	err := ticdc.destClusterCli.List(ctx, podList, client.InNamespace(tc.Namespace), client.MatchingLabels(map[string]string{
		label.ComponentLabelKey: string(cloudv1.ComponentTiCDC),
		label.InstanceLabelKey:  tc.Name,
	}))
	if client.IgnoreNotFound(err) != nil {
		return nil, err
	}

	if tc.Spec.TiCDC.Replicas != int32(len(podList.Items)) {
		return nil, fmt.Errorf("pod num %d dose not equals to tc replicas %d", len(podList.Items), tc.Spec.TiCDC.Replicas)
	}

	topology := make(map[string]int)
	for _, pod := range podList.Items {
		if !ticdc.checkPodBelongsToTheTiCDC(cr, &pod) {
			continue
		}

		if pod.Spec.NodeName == "" {
			return nil, fmt.Errorf("not all pod schedule on node")
		}

		node := &corev1.Node{}
		err = ticdc.destClusterCli.Get(ctx, client.ObjectKey{Name: pod.Spec.NodeName}, node)
		if err != nil {
			return nil, err
		}
		num, exist := topology[node.Labels[TiCDCZoneLabel]]
		if !exist {
			topology[node.Labels[TiCDCZoneLabel]] = 1
		} else {
			topology[node.Labels[TiCDCZoneLabel]] = num + 1
		}
	}
	ticdc.log.Info("get topology from pod", "topology", topology)

	// replica = 1， need complement zero node number zone
	if len(topology) < DefaultTiCDCZoneNum {
		ngTopo, err := ticdc.describeNodegroupTopology(ctx, cr, &NgFilter{
			SpecifyLabels: map[string]string{
				TiCDCCrIdLabel:    string(cr.UID),
				TiCDCClusterLabel: *cr.Spec.CDCClusterID,
				common.Component:  string(cloudv1.ComponentTiCDC),
			},
		})
		if err != nil {
			return nil, err
		}
		for zone := range ngTopo {
			if _, exist := topology[zone]; !exist {
				topology[zone] = 0
				break
			}
		}
		ticdc.log.Info("get topology from pod and node group", "topology", topology)
	}

	return topology, nil
}

func (ticdc *NewArchTiCDCCluster) ensureTiCDCReplicasReady(tc *v1alpha1.TidbCluster) bool {
	tcSpecReplicas := tc.Spec.TiCDC.Replicas
	tcReadyReplicas := int32(0)
	if tc.Status.TiCDC.StatefulSet != nil {
		tcReadyReplicas = tc.Status.TiCDC.StatefulSet.ReadyReplicas
	}

	return tcSpecReplicas == tcReadyReplicas
}

// if there are more than one ticdc cluster under one tidb namespace, we need filter pods
func (ticdc *NewArchTiCDCCluster) checkPodBelongsToTheTiCDC(cr *cloudv1.TiCDC, pod *corev1.Pod) bool {
	crID, exist := pod.Spec.NodeSelector[TiCDCCrIdLabel]
	if exist && crID == string(cr.UID) {
		return true
	}
	return false
}