package cdc

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

	"github.com/coreos/go-semver/semver"
	"github.com/go-logr/logr"
	certmanagerv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
	"github.com/jetstack/cert-manager/pkg/apis/certmanager/v1alpha2"
	cmmeta "github.com/jetstack/cert-manager/pkg/apis/meta/v1"
	"github.com/pelletier/go-toml"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/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/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"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"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	v1 "github.com/tidbcloud/infra-provider/pkg/pingcap/apis/v1"
	"github.com/tidbcloud/infra-provider/pkg/pingcap/label"
	"github.com/tidbcloud/infra-provider/pkg/provider"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

var (
	versionHash = regexp.MustCompile("-[0-9]+-g[0-9a-f]{7,}(-dev)?")
	errNotTiCDC = errors.New("not ticdc resource")
)

// TiCDCReconciler reconciles a TiCDC object
type TiCDCReconciler struct {
	reconciler *reconciler.Reconciler
	Log        logr.Logger
}

type ticdcConnector struct {
	client    client.Client
	apiReader client.Reader

	// client to the target cluster, only used for the shared K8s cluster now.
	// if this client is set, no need to try to get a client for a shoot (because we are running for a non-gardener based cluster)
	clusterCli client.Client
	event      record.EventRecorder
	log        logr.Logger
	cdc        provider.Cdc
}

// Connect initializes the ticdcConnector, there are three cases that where ticdc is deployed
//  1. deploy in upstream cluster, within the upstream cluster namespace,
//     there is one `TidbCluster` CR with ticdc
//  2. deploy in downstream and downstream exists, use the downstream namespace,
//     there are two `TidbCluster` CR, one is the downstream and another is an empty cluster only with ticdc cluster
//  3. deploy in downstream and downstream not exists, use the upstream namesapce + "ticdc" appendix
//     there is a `TidbCluster` only with ticdc
func (dc ticdcConnector) Connect(ctx context.Context, obj client.Object) (reconciler.ExternalObjClient, error) {
	d, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return nil, errNotTiCDC
	}

	var (
		upstreamCluster *cloudv1.Cluster
		// destCluster is nil when the `ClusterName` is nil,
		// that means we deploy a ticdc cluster to a region that has no tidb cluster
		destCluster *cloudv1.Cluster
	)
	if !isDRTiCDC(d) && d.Spec.ClusterName == "" {
		return nil, errors.New("missing ClusterName")
	}

	// if the ClusterName is set, deploy cluster into `ClusterName`
	if d.Spec.ClusterName != "" {
		var cluster cloudv1.Cluster
		namespacedName := types.NamespacedName{
			Name:      d.Spec.ClusterName,
			Namespace: d.ObjectMeta.Namespace,
		}
		if err := dc.client.Get(ctx, namespacedName, &cluster); err != nil {
			return nil, errors.Annotatef(err, "get cluster with ClusterName:%s", d.Spec.ClusterName)
		}
		destCluster = &cluster
	}
	// it's normal case, cdc cluster is deployed in upstream tidb cluster
	if !isDRTiCDC(d) {
		upstreamCluster = destCluster
	}
	// find the destination network ticdc deployed to
	destNetwork, err := dc.getDestNetwork(ctx, d.ObjectMeta.Namespace, d.Spec.NetworkName)
	if err != nil {
		return nil, errors.AddStack(err)
	}

	networkProvider := new(cloudv1.Provider)
	err = dc.client.Get(ctx, types.NamespacedName{Name: destNetwork.Spec.Provider}, networkProvider)
	if err != nil {
		return nil, errors.Annotate(err, "get provider")
	}

	providerType := d.Labels[cloudv1.ProviderTypeLabel]

	var destClusterCli client.Client
	// TODO: remove switch statement, all provider should be implement cdc interface
	switch providerType {
	case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		// this should already be set when initialized the `ticdcConnector`
		destClusterCli = dc.clusterCli

	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		c, err := dc.cdc.GetCdcClient(ctx, destNetwork)
		if err != nil {
			return nil, errors.AddStack(err)
		}
		destClusterCli = c
	default:
		return nil, errors.Errorf("unsupported provider type")
	}

	var upstreamClusterCli client.Client
	// connect upstream cluster
	if isDRTiCDC(d) {
		var cluster cloudv1.Cluster
		if err := dc.apiReader.Get(ctx, types.NamespacedName{
			Name:      d.Spec.UpstreamClusterName,
			Namespace: d.Namespace,
		}, &cluster); err != nil {
			return nil, errors.Annotate(err, "get upstream cluster")
		}
		upstreamCluster = &cluster

		var upstreamNetwork cloudv1.Network
		networkNamespaceName := types.NamespacedName{
			Name:      upstreamCluster.Spec.NetworkName,
			Namespace: d.ObjectMeta.Namespace,
		}
		err = dc.apiReader.Get(ctx, networkNamespaceName, &upstreamNetwork)
		if err != nil {
			return nil, errors.Annotate(err, "get upstream network")
		}

		switch providerType {
		case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
			// this should already be set when initialized the `ticdcConnector`
			upstreamClusterCli = dc.clusterCli
		case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
			upstreamClusterCli = nil // no need get upstream k8s connection
		}
	} else {
		upstreamClusterCli = destClusterCli
	}
	return BuildTiCDCCluster(providerType, &ticdcCluster{
		destClusterCli:                   destClusterCli,
		cli:                              dc.client,
		provider:                         networkProvider,
		event:                            dc.event,
		providerType:                     providerType,
		cdc:                              dc.cdc,
		log:                              dc.log,
		upstreamCluster:                  upstreamCluster,
		upstreamClusterCli:               upstreamClusterCli,
		referenceOwner:                   getReferenceOwner(upstreamCluster, destCluster),
		destNetwork:                      destNetwork,
		destNamespace:                    getDestNamespace(upstreamCluster, destCluster),
		nodeGroupNamePrefix:              ticdcNdeGroupNamePrefix(d, destCluster),
		destCluster:                      destCluster,
		crossRegionClusterNameLabelValue: getCrossRegionTiDBClusterName(destCluster),
		coreDNSName:                      corednsName(d, destCluster),
	})
}

func BuildTiCDCCluster(provider string, cls *ticdcCluster) (reconciler.ExternalObjClient, error) {
	if cls == nil {
		return nil, errors.New("empty ticdc cluster")
	}
	return BuildNewArchTiCDCCluster(*cls), nil
}

func (dc ticdcConnector) getDestNetwork(ctx context.Context, namespace, networkName string) (*cloudv1.Network, error) {
	var destNetwork cloudv1.Network
	networkNamespaceName := types.NamespacedName{
		Name:      networkName,
		Namespace: namespace,
	}
	err := dc.client.Get(ctx, networkNamespaceName, &destNetwork)
	if err != nil {
		return nil, errors.Annotate(err, "get network")
	}
	return &destNetwork, nil
}

type ticdcCluster struct {
	cli client.Client

	providerType string

	provider            *cloudv1.Provider
	event               record.EventRecorder
	cdc                 provider.Cdc
	log                 logr.Logger
	referenceOwner      *cloudv1.Cluster
	nodeGroupNamePrefix string
	// upstreamClusterCli to create a LB in upstream tidb cluster if the ticdc is deployed in downstream
	upstreamClusterCli client.Client
	upstreamCluster    *cloudv1.Cluster

	// crossRegionClusterNameLabelValue is used by node scaler to find the cross region cluster
	crossRegionClusterNameLabelValue string
	destNamespace                    string

	destClusterCli client.Client
	destNetwork    *cloudv1.Network
	destCluster    *cloudv1.Cluster
	coreDNSName    string
}

// ticdc only have Creating/Running/Scaling/Upgrading/Deleting/Unavailable status now
// even if only one TiCDC is Running, the cluster is available
func (dc *ticdcCluster) Observe(ctx context.Context, obj client.Object) (reconciler.ExternalObservation, error) {
	d, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return reconciler.ExternalObservation{}, errNotTiCDC
	}
	if err := dc.setOwnerReference(ctx, d); err != nil {
		return reconciler.ExternalObservation{}, errors.Annotate(err, "set owner reference")
	}

	var tc v1alpha1.TidbCluster

	if err := dc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(d),
		Namespace: dc.destNamespace,
	}, &tc); err != nil {
		if isDRTiCDC(d) && apierrors.IsNotFound(err) {
			return reconciler.ExternalObservation{ResourceExists: false, ResourceUpToDate: false}, nil
		}
		return reconciler.ExternalObservation{}, errors.Annotate(err, "get tc failed")
	}
	// if tc.Spec.TiCDC is nil, create it
	if tc.Spec.TiCDC == nil {
		err := dc.getTiCDCAsts(ctx, tc.Name, tc.Namespace)
		if apierrors.IsNotFound(err) {
			return reconciler.ExternalObservation{ResourceExists: false, ResourceUpToDate: false}, nil
		}
		if err != nil {
			return reconciler.ExternalObservation{}, errors.Annotate(err, "get ticdc asts failed")
		}
	}
	url, err := dc.cdc.GetNetworkProxyBaseURL(ctx, dc.destNetwork)
	if err != nil {
		return reconciler.ExternalObservation{}, errors.Annotate(err, "failed to get base proxy url")
	}
	// cluster proxy will add tidb prefix as the namespace, so we trim the prefix here
	cdcBaseProxyUrl := fmt.Sprintf("%s/proxy/cluster/%s", url, strings.TrimPrefix(dc.destNamespace, "tidb"))
	if cdcBaseProxyUrl != d.Status.BaseProxyURL {
		newCDC := d.DeepCopy()
		newCDC.Status.BaseProxyURL = cdcBaseProxyUrl
		err := dc.cli.Status().Update(ctx, newCDC)
		if err != nil {
			return reconciler.ExternalObservation{}, err
		}
	}
	// renew the status of ticdc cluster
	// if ticdc cluster status is `creating` ignore for the creating is the initialize status of ticdc cluster
	status := dc.computeTicdcStatus(&tc, d)

	if status != d.Status.Phase {
		newCDC := d.DeepCopy()
		newCDC.Status.Phase = status
		err := dc.cli.Status().Update(ctx, newCDC)
		if err != nil {
			return reconciler.ExternalObservation{}, err
		}
		dc.event.Eventf(obj, corev1.EventTypeNormal, fmt.Sprintf("CDC%s", status), fmt.Sprintf("CDC %s", status))
		if status == cloudv1.TiCDCPhaseRunning {
			return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, nil
		}
	}

	if dc.providerType == consts.ProviderTypeEKS || dc.providerType == consts.ProviderTypeGKE {
		if _, exist := d.Annotations[TiCDCResourceKeyLabel]; !exist {
			return reconciler.ExternalObservation{ResourceExists: false, ResourceUpToDate: false}, nil
		}
	}

	return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false}, nil
}

func (dc *ticdcCluster) setOwnerReference(ctx context.Context, d *cloudv1.TiCDC) error {
	if !dc.cdc.IsDPCompatMode() {
		if d.OwnerReferences == nil {
			err := controllerutil.SetOwnerReference(dc.referenceOwner, d, dc.cli.Scheme())
			if err != nil {
				return err
			}
			return dc.cli.Update(ctx, d)
		}
	}
	return nil
}

func (dc *ticdcCluster) sync(ctx context.Context, d *cloudv1.TiCDC, labels map[string]string) (controllerutil.OperationResult, error) {
	var corednsClusterIP string
	var coreDNSIps string
	var err error

	// make dns resolver worker
	if isDRTiCDC(d) {
		coreDNSIps, err = dc.ensureUpstreamCoreDNS(ctx, d)
		if err != nil {
			return controllerutil.OperationResultNone, err
		}

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

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

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

	tc := new(v1alpha1.TidbCluster)
	if err := dc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(d),
		Namespace: dc.destNamespace,
	}, tc); err != nil {
		if isDRTiCDC(d) && apierrors.IsNotFound(err) {
			// delete pvc when ticdc scale-in
			pvDeletePolicy := corev1.PersistentVolumeReclaimDelete
			// TODB: add DNS Config
			tc = &v1alpha1.TidbCluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      getTiDBClusterNameForTiCDC(d),
					Namespace: dc.destNamespace,
					// node scaler require the labels to query or build TiCDC object
					Labels: map[string]string{
						consts.ClusterTicdcLabelKey:        d.Name,
						consts.ClusterNamespaceLabelKey:    d.Namespace,
						consts.ClusterNameLabelKey:         dc.crossRegionClusterNameLabelValue,
						consts.ClusterNetworkLabelKey:      dc.destNetwork.Name,
						consts.UpstreamClusterNameLabelKey: d.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" + d.Spec.UpstreamClusterName,
						Name:      utils.ClusterPrefix,
					},
				},
			}
		} else {
			return controllerutil.OperationResultNone, errors.Annotate(err, "get tc failed")
		}
	}

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

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

	if result, err := dc.createTiCDCComponentCert(ctx, tc.Namespace, tc.Name, upstreamNamespace, utils.ClusterPrefix); err != nil {
		dc.event.Eventf(d, corev1.EventTypeWarning, "FailedCreateTiCDCComponentCert", "create cdc tls cert failed")
		return controllerutil.OperationResultNone, errors.Annotate(err, "create cdc tls cert failed")
	} else if result == controllerutil.OperationResultCreated {
		dc.event.Eventf(d, corev1.EventTypeNormal, "CreateTiCDCComponentCert", "Create cdc tls cert successful")
	}
	result, err := controllerutil.CreateOrUpdate(ctx, dc.destClusterCli, tc, func() error {
		if updateErr := dc.overlayTicdc(tc, d, labels, nil, &gitopsv2.Gitops{Cli: dc.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")
	}
	return result, nil
}

func (dc *ticdcCluster) prepareNodeGroup(ctx context.Context, d *cloudv1.TiCDC) (map[string]string, error) {
	name := dc.nodeGroupNamePrefix
	zones, err := dc.provider.FindRegionZones(dc.destNetwork.Spec.Region)
	if err != nil {
		return nil, err
	}
	set := &nodegroup.Set{
		Name: name,
		NodeGroups: []cloudv1.NodegroupSpec{
			{
				Name:   name,
				Labels: map[string]string{consts.GardenerWorkerPrefixLabelName: name},
				MachineSpec: &cloudv1.MachineSpec{
					MachineType: d.Spec.InstanceType,
				},
				Taints:   utils.Taint(cloudv1.ComponentTiCDC),
				Maximum:  int32(d.Spec.Replicas),
				Minimum:  int32(d.Spec.Replicas),
				Zones:    zones,
				NodeRole: d.Spec.NodeRole,
			},
		},
	}
	// infra-api manage cdc nodes by node controller, so we don't need to create node group. node controller will update shoot to create node group
	// eks provider manage cdc nodes by crossplane nodegroups resources, so we need to create node groups in controller.
	return dc.cdc.CreateCdcNodeGroup(ctx, d, dc.destCluster, set)
}

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

	labels, err := dc.prepareNodeGroup(ctx, d)
	if err != nil {
		return err
	}
	result, err := dc.sync(ctx, d, labels)
	if err != nil {
		return err
	}
	if result != controllerutil.OperationResultNone {
		dc.event.Eventf(obj, corev1.EventTypeNormal, "CDCCreated", "Create cdc in shoot successful")
		d.Status.Phase = cloudv1.TiCDCPhaseCreating

		return dc.cli.Status().Update(ctx, d)
	}

	return nil
}

func (dc *ticdcCluster) Delete(ctx context.Context, obj client.Object) (reconciler.ExternalDeletion, error) {
	tc := new(v1alpha1.TidbCluster)
	d, ok := obj.(*cloudv1.TiCDC)
	if !ok {
		return reconciler.ExternalDeletion{}, errNotTiCDC
	}
	err := dc.getTicdcCluster(ctx, tc, d)
	// 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(d) {
		// 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 = dc.destClusterCli.Update(ctx, tc); err != nil && client.IgnoreNotFound(err) != nil {
				dc.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 = dc.getTicdcCluster(ctx, tc, d)
		// tidbcluster already deleted, we can think cdc also removed
		if apierrors.IsNotFound(err) {
			return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
		}

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

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

	err = dc.deleteTiCDCComponentCert(ctx, tc.Namespace, tc.Name)
	if client.IgnoreNotFound(err) != nil {
		dc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDCComponentCert", "failed delete cdc component cert in shoot cluster")
		return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc component cert")
	}

	set := &nodegroup.Set{
		Name: dc.nodeGroupNamePrefix,
	}
	err = dc.cdc.DeleteCdcNodeGroup(ctx, d, set)
	if client.IgnoreNotFound(err) != nil {
		dc.event.Eventf(obj, corev1.EventTypeWarning, "FailedDeleteCDCNodes", "failed delete cdc nodes")
		return reconciler.ExternalDeletion{ResourceDeleted: false}, errors.Annotate(err, "remove ticdc node")
	}

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

	dc.event.Eventf(obj, corev1.EventTypeNormal, "CDCDeleted", "Delete cdc successful")

	return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
}

func (dc *ticdcCluster) ensureTiCDCPodClear(ctx context.Context, tc *v1alpha1.TidbCluster) error {
	podList := &corev1.PodList{}
	err := dc.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 len(podList.Items) > 0 {
		return fmt.Errorf("ensure ticdc pod clear failed, still has %d pod alive", len(podList.Items))
	}
	return nil
}

func (dc *ticdcCluster) getTicdcCluster(ctx context.Context, tc *v1alpha1.TidbCluster, cdc *cloudv1.TiCDC) error {
	err := dc.destClusterCli.Get(ctx, types.NamespacedName{
		Name:      getTiDBClusterNameForTiCDC(cdc),
		Namespace: dc.destNamespace,
	}, tc)
	return err
}

func (dc *ticdcCluster) deleteTicdcPvc(ctx context.Context, ns string) error {
	pvcs := new(corev1.PersistentVolumeClaimList)
	cdcLabelSelector, err := labels.Parse("app.kubernetes.io/component=ticdc")
	if err != nil {
		return err
	}
	err = dc.destClusterCli.List(ctx, pvcs, &client.ListOptions{Namespace: ns, LabelSelector: cdcLabelSelector})
	if err != nil {
		return err
	}
	for _, pvc := range pvcs.Items {
		if pvc.Spec.StorageClassName != nil {
			err = dc.destClusterCli.Delete(ctx, &pvc)
			if client.IgnoreNotFound(err) != nil {
				return err
			}
		}
	}
	return nil
}

func (dc *ticdcCluster) deleteTiCDCAsts(ctx context.Context, tcName string, ns string) error {
	comp := cloudv1.ComponentTiCDC
	cdcAsts := &v1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s", tcName, comp),
			Namespace: ns,
		},
	}
	err := dc.destClusterCli.Delete(ctx, cdcAsts)
	return client.IgnoreNotFound(err)
}

func (dc *ticdcCluster) getTiCDCAsts(ctx context.Context, tcName string, ns string) error {
	comp := cloudv1.ComponentTiCDC
	cdcAsts := new(v1.StatefulSet)

	key := types.NamespacedName{
		Name:      fmt.Sprintf("%s-%s", tcName, comp),
		Namespace: ns,
	}
	return dc.destClusterCli.Get(ctx, key, cdcAsts)
}

func (dc *ticdcCluster) Update(_ context.Context, _ client.Object) error {
	return nil
}

// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=ticdcs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=ticdcs/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=ticdcs/finalizers,verbs=update
func (r *TiCDCReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	start := time.Now()
	defer func() {
		metricsctrl.ReconcileDuration.WithLabelValues("ticdc").Observe(time.Since(start).Seconds())
	}()
	logger := r.Log.WithValues("ticdc", req.NamespacedName)
	logger.Info("start reconcile ticdc")
	ctx = log.IntoContext(ctx, logger)
	return r.reconciler.Reconcile(ctx, req, TiCDCFinalizer)
}

func NewTiCDCReconcilerCompat(infraCli client.Client, infraApiReader client.Reader, log logr.Logger, clusterCli client.Client, event record.EventRecorder, cdc provider.Cdc) *TiCDCReconciler {
	return &TiCDCReconciler{
		Log: log,
		reconciler: &reconciler.Reconciler{
			Client: infraCli,
			ExternalObj: ticdcConnector{
				client:     infraCli,
				apiReader:  infraApiReader,
				clusterCli: clusterCli,
				event:      event,
				log:        log,
				cdc:        cdc,
			},
			NewResource: func() client.Object {
				return &cloudv1.TiCDC{}
			},
		},
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *TiCDCReconciler) SetupWithManager(mgr ctrl.Manager, p ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.TiCDC{}, builder.WithPredicates(p...)).
		WithOptions(controller.Options{RateLimiter: common.DefaultRateLimiter(), MaxConcurrentReconciles: 10}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)
}

// the Creating means the tc.Status.TiCDC.Syned is `false`
// the ticdc is Deleting when the tc.Spec.TiCDC.Replicas equals 0 and the tc phase is Scale or tc.Spec.TiCDC is nil
// the Unavailable status means the tc.Spec.TiCDC.Replicas > 0 and the tc.Status.TiCDC.StatuefulSet.ReadyReplicas = 0
func (dc *ticdcCluster) computeTicdcStatus(tc *v1alpha1.TidbCluster, cr *cloudv1.TiCDC) cloudv1.TiCDCPhase {
	key, exist := cr.Annotations[TiCDCResourceKeyLabel]
	switch tc.Status.TiCDC.Phase {
	case v1alpha1.NormalPhase:
		// If last status is normal, but ready replicas to zero, maybe ticdc service down.
		if tc.Status.TiCDC.StatefulSet.ReadyReplicas == 0 && cr.Status.Phase == cloudv1.TiCDCPhaseRunning {
			return cloudv1.TiCDCPhaseUnavailable
		}
		if exist {
			// use resourceKey to ensure ticdc is running or scaling
			if instanceType, replicas, err := TiCDCResourceValue(key); err == nil {
				if instanceType == cr.Spec.InstanceType && replicas == cr.Spec.Replicas {
					return cloudv1.TiCDCPhaseRunning
				} else {
					return cloudv1.TiCDCPhaseScaling
				}
			}
		} else {
			if tc.Status.TiCDC.StatefulSet.ReadyReplicas == int32(cr.Spec.Replicas) {
				return cloudv1.TiCDCPhaseRunning
			} else if cr.Status.Phase != "" && tc.Status.TiCDC.StatefulSet.ReadyReplicas > 0 {
				// if desire replicas not equal with ready replicas and ready replicas > 0, status is scaling
				return cloudv1.TiCDCPhaseScaling
			}
		}
		// if desire replicas equal with ready replicas and last phase is null, status is creating
		return cloudv1.TiCDCPhaseCreating
	case v1alpha1.UpgradePhase:
		return cloudv1.TiCDCPhaseUpgrading
	case v1alpha1.ScalePhase:
		if tc.Spec.TiCDC.Replicas != 0 {
			return cloudv1.TiCDCPhaseScaling
		}
		return cloudv1.TiCDCPhaseDeleting
	default:
		return cloudv1.TiCDCPhaseCreating
	}
}

// overlay ticdc cluster to tidb cluster
func (dc *ticdcCluster) overlayTicdc(tc *v1alpha1.TidbCluster,
	d *cloudv1.TiCDC,
	labels map[string]string,
	credData *AccountCredentialData,
	gitopsV2 gitopsv2.CCInterface) error {
	wantTicdc := d.Spec.Replicas > 0
	switch {
	case wantTicdc:
		var (
			config                   *v1alpha1.CDCConfigWraper
			TiCDCImage, TiCDCVersion string
			err                      error
		)
		if d.Spec.Configuration != nil {
			config, TiCDCImage, TiCDCVersion, err = dc.initCDCConfigByConfigCenter(d.Spec.Configuration)
			if err != nil {
				return errors.Annotate(err, "init cdc config by config-center failed")
			}
		} else {
			config, TiCDCImage, TiCDCVersion, err = dc.initCDCConfigByGitOps(gitopsV2)
			if err != nil {
				return errors.Annotate(err, "init cdc config by gitops failed")
			}
		}
		// cluster-id should always set by deployment instead of gitops
		if d.Spec.CDCClusterID != nil && supportClusterID(TiCDCVersion) {
			config.Set("cluster-id", *d.Spec.CDCClusterID)
		}
		// dir should always set by deployment instead of gitops
		config.Set("data-dir", TiCDCDefaultDirPath)

		// Make the cdc log print to stdout, container logs should not be output to a file, otherwise it will cause disk exhaustion.
		// Docker log drive will collect log and rotate the log.
		config.Set("log-file", "")

		// Because `GetResourceSpec` doesn't need the AWS client and ARM arch flag,
		// so it's OK to pass `nil, false` here.
		ticdcResource, err := gitops.NewHardcode().GetResourceSpec(dc.provider, d)
		if err != nil {
			return errors.Annotate(err, "get ticdc resources spec")
		}

		if tc.Labels == nil {
			tc.Labels = make(map[string]string)
		}
		// FIXME: only used for gardner
		providerType := d.Labels[cloudv1.ProviderTypeLabel]
		if providerType == consts.ProviderTypeAWS || providerType == consts.ProviderTypeGCP {
			tc.Labels[consts.ClusterTicdcLabelKey] = d.ObjectMeta.Name
			tc.Labels[consts.ClusterNetworkLabelKey] = dc.destNetwork.Name
		}
		tc.Spec.TiCDC = &v1alpha1.TiCDCSpec{
			ComponentSpec: v1alpha1.ComponentSpec{
				Affinity:           utils.Affinity(cloudv1.ComponentTiCDC),
				Tolerations:        utils.Toleration(cloudv1.ComponentTiCDC),
				HostNetwork:        pointer.Bool(true),
				PodSecurityContext: &corev1.PodSecurityContext{},
				Version:            &TiCDCVersion,
			},

			ResourceRequirements: ticdcResource.Resources,
			BaseImage:            TiCDCImage,
			Replicas:             int32(d.Spec.Replicas),
			Config:               config,
			StorageVolumes: []v1alpha1.StorageVolume{
				{
					Name:             TiCDCDefaultDirName,
					StorageClassName: &ticdcResource.StorageClass,
					StorageSize:      fmt.Sprintf("%dGi", int(d.Spec.StorageSize.Value())/1024/1024/1024),
					MountPath:        TiCDCDefaultDirPath,
				},
			},
		}
		if len(labels) > 0 {
			tc.Spec.TiCDC.ComponentSpec.NodeSelector = labels
			if (dc.providerType == consts.ProviderTypeEKS || dc.providerType == consts.ProviderTypeGKE) && d.Spec.Replicas > DefaultTiCDCZoneNum {
				tc.Spec.TiCDC.Replicas = DefaultTiCDCZoneNum // deploy ticdc DefaultTiCDCZoneNum, then scale out one by one
			}
		}
		if credData != nil {
			tc.Spec.TiCDC.ComponentSpec.AdditionalVolumes = append(tc.Spec.TiCDC.ComponentSpec.AdditionalVolumes, credData.volume)
			tc.Spec.TiCDC.ComponentSpec.AdditionalVolumeMounts = append(tc.Spec.TiCDC.ComponentSpec.AdditionalVolumeMounts, credData.mount)
			tc.Spec.TiCDC.ComponentSpec.Env = append(tc.Spec.TiCDC.ComponentSpec.Env, credData.env)
		}
	case !wantTicdc:
		// tiflash is not expected but we have tiflash in tc spec, disable it
		tc.Spec.TiCDC = nil
	}
	return nil
}

func (dc *ticdcCluster) initCDCConfigByGitOps(
	gitopsV2 gitopsv2.CCInterface,
) (cdcConf *v1alpha1.CDCConfigWraper, image, tag string, err error) {
	cdcConf = v1alpha1.NewCDCConfig()
	image = DefaultTiCDCVersion
	tag = DefaultTiCDCImage
	cc, err := gitopsV2.DescribeClusterConfiguration(dc.referenceOwner)
	if err != nil {
		return nil, "", "", errors.Annotate(err, "get cluster config failed")
	}

	if cc.Spec.TiCDC != nil {
		cdcConf.GenericConfig.MP = cc.Spec.TiCDC.Config.MP
		if cc.Spec.TiCDC.BaseImage != nil {
			image = dc.cdc.OverrideImage(dc.referenceOwner, *cc.Spec.TiCDC.BaseImage)
		}
		if cc.Spec.TiCDC.Tag != nil {
			tag = *cc.Spec.TiCDC.Tag
		}
	}
	return
}

func (dc *ticdcCluster) initCDCConfigByConfigCenter(
	conf *cloudv1.DPConfiguration,
) (cdcConf *v1alpha1.CDCConfigWraper, image, tag string, err error) {
	cdcConf = v1alpha1.NewCDCConfig()
	image = dc.cdc.OverrideImage(dc.referenceOwner, conf.BaseImage)
	tag = conf.Tag

	if conf.Config == "" {
		return
	}

	if err = toml.Unmarshal([]byte(conf.Config), &cdcConf.GenericConfig.MP); err != nil {
		return nil, "", "", err
	}
	return
}

func (dc *ticdcCluster) deleteTiCDCComponentCert(ctx context.Context, ns, clusterName string) error {
	comp := cloudv1.ComponentTiCDC
	componentCertificate := &v1alpha2.Certificate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      tidbClusterComponentSecretName(clusterName, comp),
			Namespace: ns,
		},
	}

	obj := dc.CompatibleMultipleVersionsCertManager(componentCertificate)
	err := dc.destClusterCli.Delete(ctx, obj)
	return client.IgnoreNotFound(err)
}

// createUpstreamClusterClientCert create a certificate for visit the upstream cluster. commonName should be match to upstream cluster
func (dc *ticdcCluster) createUpstreamClusterClientCert(ctx context.Context, ns, clusterName, upstreamNamespace string) (controllerutil.OperationResult, error) {
	kind, name, group, duration := dc.cdc.CertificateOptions(dc.destCluster)
	componentCertificate := &v1alpha2.Certificate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-cluster-client-secret", clusterName),
			Namespace: ns,
		},

		Spec: v1alpha2.CertificateSpec{
			SecretName: fmt.Sprintf("%s-cluster-client-secret", clusterName),
			Duration: &metav1.Duration{
				Duration: duration,
			},
			RenewBefore: &metav1.Duration{
				Duration: consts.RenewBefore,
			},
			CommonName:   util.FnvHash(upstreamNamespace),
			Organization: []string{consts.CertOrganization},
			Usages: []v1alpha2.KeyUsage{
				v1alpha2.UsageClientAuth,
			},
			IssuerRef: cmmeta.ObjectReference{
				Name:  name,
				Kind:  kind,
				Group: group,
			},
		},
	}
	obj := dc.CompatibleMultipleVersionsCertManager(componentCertificate)

	return k8s.CreateIgnoreExists(ctx, dc.destClusterCli, obj)
}

// TODO: the discovery pod does not need any certificate, remove this cert when operator fix it
func (dc *ticdcCluster) createCertForDiscovery(ctx context.Context, ns, clusterName, upstreamNamespace string) (controllerutil.OperationResult, error) {
	kind, name, group, duration := dc.cdc.CertificateOptions(dc.destCluster)
	componentCertificate := &v1alpha2.Certificate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%s-cluster-secret", clusterName, cloudv1.ComponentPD),
			Namespace: ns,
		},

		Spec: v1alpha2.CertificateSpec{
			SecretName: fmt.Sprintf("%s-%s-cluster-secret", clusterName, cloudv1.ComponentPD),
			Duration: &metav1.Duration{
				Duration: duration,
			},
			RenewBefore: &metav1.Duration{
				Duration: consts.RenewBefore,
			},
			CommonName:   util.FnvHash(upstreamNamespace),
			Organization: []string{consts.CertOrganization},
			Usages: []v1alpha2.KeyUsage{
				v1alpha2.UsageClientAuth,
			},
			IssuerRef: cmmeta.ObjectReference{
				Name:  name,
				Kind:  kind,
				Group: group,
			},
		},
	}
	obj := dc.CompatibleMultipleVersionsCertManager(componentCertificate)

	return k8s.CreateIgnoreExists(ctx, dc.destClusterCli, obj)
}

// TODO: duplicate with cluster, merge it
func (dc *ticdcCluster) createTiCDCComponentCert(ctx context.Context, ns, clusterName, upstreamNamespace, upstreamClusterName string) (controllerutil.OperationResult, error) {
	comp := cloudv1.ComponentTiCDC
	commonName := ns
	componentDNSNames := []string{
		fmt.Sprintf("%s-%s", clusterName, comp),
		fmt.Sprintf("%s-%s.%s", clusterName, comp, ns),
		fmt.Sprintf("%s-%s.%s.svc", clusterName, comp, ns),
		fmt.Sprintf("%s-%s-peer", clusterName, comp),
		fmt.Sprintf("%s-%s-peer.%s", clusterName, comp, ns),
		fmt.Sprintf("%s-%s-peer.%s.svc", clusterName, comp, ns),
		fmt.Sprintf("*.%s-%s-peer", clusterName, comp),
		fmt.Sprintf("*.%s-%s-peer.%s", clusterName, comp, ns),
		fmt.Sprintf("*.%s-%s-peer.%s.svc", clusterName, comp, ns),
	}
	if upstreamNamespace != "" {
		commonName = upstreamNamespace
		upstreamComponentDNSNames := []string{
			fmt.Sprintf("%s-%s", upstreamClusterName, comp),
			fmt.Sprintf("%s-%s.%s", upstreamClusterName, comp, upstreamNamespace),
			fmt.Sprintf("%s-%s.%s.svc", upstreamClusterName, comp, upstreamNamespace),
			fmt.Sprintf("%s-%s-peer", upstreamClusterName, comp),
			fmt.Sprintf("%s-%s-peer.%s", upstreamClusterName, comp, upstreamNamespace),
			fmt.Sprintf("%s-%s-peer.%s.svc", upstreamClusterName, comp, upstreamNamespace),
			fmt.Sprintf("*.%s-%s-peer", upstreamClusterName, comp),
			fmt.Sprintf("*.%s-%s-peer.%s", upstreamClusterName, comp, upstreamNamespace),
			fmt.Sprintf("*.%s-%s-peer.%s.svc", upstreamClusterName, comp, upstreamNamespace),
		}
		componentDNSNames = append(componentDNSNames, upstreamComponentDNSNames...)
	}

	kind, name, group, duration := dc.cdc.CertificateOptions(dc.destCluster)

	componentCertificate := &v1alpha2.Certificate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      tidbClusterComponentSecretName(clusterName, comp),
			Namespace: ns,
		},
		Spec: v1alpha2.CertificateSpec{
			SecretName: tidbClusterComponentSecretName(clusterName, comp),
			Duration: &metav1.Duration{
				Duration: duration,
			},
			RenewBefore: &metav1.Duration{
				Duration: consts.RenewBefore,
			},
			CommonName:   util.FnvHash(commonName),
			Organization: []string{consts.CertOrganization},
			Usages: []v1alpha2.KeyUsage{
				v1alpha2.UsageServerAuth,
				v1alpha2.UsageClientAuth,
			},
			DNSNames: componentDNSNames,
			IPAddresses: []string{
				"127.0.0.1",
				"::1",
			},
			IssuerRef: cmmeta.ObjectReference{
				Name:  name,
				Kind:  kind,
				Group: group,
			},
		},
	}

	obj := dc.CompatibleMultipleVersionsCertManager(componentCertificate)

	return k8s.CreateIgnoreExists(ctx, dc.destClusterCli, obj)
}

func tidbClusterComponentSecretName(clusterName string, comp cloudv1.ComponentType) string {
	return fmt.Sprintf("%s-%s-cluster-secret", clusterName, comp)
}

func getTiDBClusterNameForTiCDC(cdc *cloudv1.TiCDC) string {
	var tcName string
	if isDRTiCDC(cdc) {
		tcName = fmt.Sprintf("%s-cdc", utils.ClusterPrefix)
	} else {
		tcName = utils.ClusterPrefix
	}
	return tcName
}

func isDRTiCDC(cdc *cloudv1.TiCDC) bool {
	return cdc.Spec.UpstreamClusterName != ""
}

func destroyCoredns(ctx context.Context, cli client.Client, name string) error {
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
	}
	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
	}
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
	}

	var objs = []client.Object{cm, deploy, svc}

	for _, obj := range objs {
		err := cli.Delete(ctx, obj)
		if client.IgnoreNotFound(err) != nil {
			return err
		}
	}

	return nil
}

func (dc *ticdcCluster) ensureUpstreamCoreDNS(ctx context.Context, d *cloudv1.TiCDC) (string, error) {
	var ipStr string
	if d.Spec.UpstreamCoreDNSIp == "" {
		var corednsPort int32 = 8053
		if dc.providerType == consts.ProviderTypeEKS || dc.providerType == consts.ProviderTypeGKE {
			//corednsPort = 53
			return "", fmt.Errorf("ticdc UpstreamCoreDNSIp is empty")
		}
		// create an internal coredns load balancer to upstream
		corednsSvc := corev1.Service{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "coredns-lb",
				Namespace: "kube-system",
			},
			Spec: corev1.ServiceSpec{
				Ports: []corev1.ServicePort{
					{
						Name: "dns",
						Port: 53,
						TargetPort: intstr.IntOrString{
							IntVal: corednsPort,
							Type:   intstr.Int,
						},
						Protocol: corev1.ProtocolTCP,
					},
				},
				Selector: map[string]string{
					"k8s-app": "kube-dns",
				},
				LoadBalancerSourceRanges: []string{"0.0.0.0/0"},
				Type:                     corev1.ServiceTypeLoadBalancer,
			},
		}
		if err := dc.upstreamClusterCli.Get(ctx, types.NamespacedName{
			Name:      corednsSvc.Name,
			Namespace: corednsSvc.Namespace,
		}, &corednsSvc); err != nil {
			if apierrors.IsNotFound(err) {
				err := dc.upstreamClusterCli.Create(ctx, &corednsSvc)
				if err != nil {
					return "", err
				}
				return "", fmt.Errorf("Wait coredns loadbalancer ready")
			} else {
				return "", err
			}
		}

		if len(corednsSvc.Status.LoadBalancer.Ingress) == 0 {
			return "", fmt.Errorf("Wait coredns loadbalancer ready")
		}

		switch dc.providerType {
		case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
			hostname := corednsSvc.Status.LoadBalancer.Ingress[0].Hostname
			ips, err := net.LookupIP(hostname)
			if err != nil {
				return "", err
			}
			var items []string
			for _, ip := range ips {
				items = append(items, ip.String())
			}
			ipStr = strings.Join(items, " ")
		case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
			ipStr = corednsSvc.Status.LoadBalancer.Ingress[0].IP
		}
	} else {
		ipStr = d.Spec.UpstreamCoreDNSIp
	}
	return ipStr, nil
}

func deployKubeDNS(ctx context.Context, cli client.Client, name, upstreamClusterName, upsreamCoreDNSLB string) error {
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
		Data: map[string]string{
			"stubDomains": fmt.Sprintf(kubeDNSConf, upstreamClusterName, upsreamCoreDNSLB),
		},
	}
	_, err := k8s.CreateIgnoreExists(ctx, cli, cm)
	if err != nil {
		return err
	}

	labels := map[string]string{
		"app":  "cdc-dns-forward",
		"name": name,
	}

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
		Spec: corev1.ServiceSpec{
			Selector: labels,
			Type:     corev1.ServiceTypeClusterIP,
			Ports: []corev1.ServicePort{
				{
					Name:     "dns-udp",
					Protocol: corev1.ProtocolUDP,
					TargetPort: intstr.IntOrString{
						IntVal: 53,
						Type:   intstr.Int,
					},
					Port: 53,
				},
				{
					Name:     "metrics",
					Protocol: corev1.ProtocolTCP,
					Port:     9153,
					TargetPort: intstr.IntOrString{
						IntVal: 9153,
						Type:   intstr.Int,
					},
				},
			},
		},
	}
	_, err = k8s.CreateIgnoreExists(ctx, cli, svc)
	if err != nil {
		return err
	}

	falseVar := false
	trueVar := true
	var runUser int64 = 1001
	var terminationGracePeriodSeconds int64 = 30
	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
			Annotations: map[string]string{
				"deployment.kubernetes.io/revision": "1",
				"k8s-app":                           "kube-dns",
			},
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: pointer.Int32(2),
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},
			Strategy: appsv1.DeploymentStrategy{
				Type: appsv1.RollingUpdateDeploymentStrategyType,
				RollingUpdate: &appsv1.RollingUpdateDeployment{
					MaxUnavailable: &intstr.IntOrString{
						Type:   intstr.Int,
						IntVal: 0,
					},
					MaxSurge: &intstr.IntOrString{
						Type:   intstr.String,
						StrVal: "10%",
					},
				},
			},

			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labels,
				},
			},
		},
	}

	_, err = controllerutil.CreateOrUpdate(ctx, cli, deploy, func() error {
		deploy.Spec.Template.Spec = corev1.PodSpec{
			DNSPolicy:                     corev1.DNSDefault,
			RestartPolicy:                 corev1.RestartPolicyAlways,
			SchedulerName:                 corev1.DefaultSchedulerName,
			ServiceAccountName:            "kube-dns",
			TerminationGracePeriodSeconds: &terminationGracePeriodSeconds,
			Tolerations: []corev1.Toleration{
				{
					Key:      "CriticalAddonsOnly",
					Operator: corev1.TolerationOpExists,
				},
			},
			Volumes: []corev1.Volume{
				{
					Name: "kube-dns-config",
					VolumeSource: corev1.VolumeSource{
						ConfigMap: &corev1.ConfigMapVolumeSource{
							DefaultMode: pointer.Int32(420),
							Optional:    &trueVar,
							LocalObjectReference: corev1.LocalObjectReference{
								Name: cm.Name,
							},
						},
					},
				},
			},
		}

		containers := []corev1.Container{
			{
				Name:  "kubedns",
				Image: "gke.gcr.io/k8s-dns-kube-dns:1.22.12-gke.0",
				Resources: corev1.ResourceRequirements{
					Limits: map[corev1.ResourceName]resource.Quantity{
						corev1.ResourceMemory: resource.MustParse("170Mi"),
					},
					Requests: map[corev1.ResourceName]resource.Quantity{
						corev1.ResourceCPU:    resource.MustParse("100m"),
						corev1.ResourceMemory: resource.MustParse("70Mi"),
					},
				},
				LivenessProbe: &corev1.Probe{
					ProbeHandler: corev1.ProbeHandler{
						HTTPGet: &corev1.HTTPGetAction{
							Path: "/healthcheck/kubedns",
							Port: intstr.IntOrString{
								Type:   intstr.Int,
								IntVal: 10054,
							},
							Scheme: corev1.URISchemeHTTP,
						},
					},
					InitialDelaySeconds: 60,
					TimeoutSeconds:      5,
					SuccessThreshold:    1,
					FailureThreshold:    5,
				},
				ReadinessProbe: &corev1.Probe{
					ProbeHandler: corev1.ProbeHandler{
						HTTPGet: &corev1.HTTPGetAction{
							Path: "/readiness",
							Port: intstr.IntOrString{
								Type:   intstr.Int,
								IntVal: 8081,
							},
							Scheme: corev1.URISchemeHTTP,
						},
					},
					InitialDelaySeconds: 3,
					TimeoutSeconds:      5,
				},
				Args: []string{
					"--domain=cluster.local.",
					"--dns-port=10053",
					"--config-dir=/kube-dns-config",
					"--v=2",
				},
				Env: []corev1.EnvVar{
					{
						Name:  "PROMETHEUS_PORT",
						Value: "10055",
					},
				},
				Ports: []corev1.ContainerPort{
					{
						Name:          "dns-local",
						Protocol:      corev1.ProtocolUDP,
						ContainerPort: 10053,
					},
					{
						Name:          "dns-tcp-local",
						Protocol:      corev1.ProtocolTCP,
						ContainerPort: 10053,
					},
					{
						Name:          "metrics",
						Protocol:      corev1.ProtocolTCP,
						ContainerPort: 10055,
					},
				},
				VolumeMounts: []corev1.VolumeMount{
					{
						Name:      "kube-dns-config",
						MountPath: "/kube-dns-config",
					},
				},
				SecurityContext: &corev1.SecurityContext{
					AllowPrivilegeEscalation: &falseVar,
					ReadOnlyRootFilesystem:   &trueVar,
					RunAsUser:                &runUser,
					RunAsGroup:               &runUser,
				},
			},
			{
				Name:  "dnsmasq",
				Image: "gke.gcr.io/k8s-dns-dnsmasq-nanny:1.22.12-gke.0",
				Resources: corev1.ResourceRequirements{
					Requests: map[corev1.ResourceName]resource.Quantity{
						corev1.ResourceCPU:    resource.MustParse("150m"),
						corev1.ResourceMemory: resource.MustParse("20Mi"),
					},
				},
				LivenessProbe: &corev1.Probe{
					ProbeHandler: corev1.ProbeHandler{
						HTTPGet: &corev1.HTTPGetAction{
							Path: "/healthcheck/dnsmasq",
							Port: intstr.IntOrString{
								Type:   intstr.Int,
								IntVal: 10054,
							},
							Scheme: corev1.URISchemeHTTP,
						},
					},
					InitialDelaySeconds: 60,
					TimeoutSeconds:      5,
					SuccessThreshold:    1,
					FailureThreshold:    5,
				},
				Args: []string{
					"-v=2",
					"-logtostderr",
					"-configDir=/etc/k8s/dns/dnsmasq-nanny",
					"-restartDnsmasq=true",
					"--",
					"-k",
					"--cache-size=1000",
					"--no-negcache",
					"--dns-forward-max=1500",
					"--log-facility=-",
					"--server=/cluster.local/127.0.0.1#10053",
					"--server=/in-addr.arpa/127.0.0.1#10053",
					"--server=/ip6.arpa/127.0.0.1#10053",
				},
				Ports: []corev1.ContainerPort{
					{
						Name:          "dns",
						Protocol:      corev1.ProtocolUDP,
						ContainerPort: 53,
					},
					{
						Name:          "dns-tcp",
						Protocol:      corev1.ProtocolTCP,
						ContainerPort: 53,
					},
				},
				VolumeMounts: []corev1.VolumeMount{
					{
						Name:      "kube-dns-config",
						MountPath: "/etc/k8s/dns/dnsmasq-nanny",
					},
				},
				SecurityContext: &corev1.SecurityContext{
					Capabilities: &corev1.Capabilities{
						Drop: []corev1.Capability{
							"all",
						},
						Add: []corev1.Capability{
							"NET_BIND_SERVICE",
							"SETGID",
						},
					},
				},
			},
			{
				Name:  "sidecar",
				Image: "gke.gcr.io/k8s-dns-sidecar:1.22.12-gke.0",
				Resources: corev1.ResourceRequirements{
					Requests: map[corev1.ResourceName]resource.Quantity{
						corev1.ResourceCPU:    resource.MustParse("10m"),
						corev1.ResourceMemory: resource.MustParse("20Mi"),
					},
				},
				LivenessProbe: &corev1.Probe{
					ProbeHandler: corev1.ProbeHandler{
						HTTPGet: &corev1.HTTPGetAction{
							Path: "/metrics",
							Port: intstr.IntOrString{
								Type:   intstr.Int,
								IntVal: 10054,
							},
							Scheme: corev1.URISchemeHTTP,
						},
					},
					InitialDelaySeconds: 60,
					TimeoutSeconds:      5,
					SuccessThreshold:    1,
					FailureThreshold:    5,
				},
				Args: []string{
					"--v=2",
					"--logtostderr",
					"--probe=kubedns,127.0.0.1:10053,kubernetes.default.svc.cluster.local,5,SRV",
					"--probe=dnsmasq,127.0.0.1:53,kubernetes.default.svc.cluster.local,5,SRV",
				},
				Ports: []corev1.ContainerPort{
					{
						Name:          "metrics",
						Protocol:      corev1.ProtocolTCP,
						ContainerPort: 10054,
					},
				},
				SecurityContext: &corev1.SecurityContext{
					AllowPrivilegeEscalation: &falseVar,
					ReadOnlyRootFilesystem:   &trueVar,
					RunAsUser:                &runUser,
					RunAsGroup:               &runUser,
				},
			},
		}
		deploy.Spec.Template.Spec.Containers = append(deploy.Spec.Template.Spec.Containers, containers...)

		return nil
	})

	return err
}

func deployCoredns(ctx context.Context, cli client.Client, name, upstreamClusterName, upsreamCoreDNSLB, providerType string) error {
	var coreDNSport int32 = 8053
	var option string = "force_tcp"
	if providerType == consts.ProviderTypeEKS || providerType == consts.ProviderTypeGKE {
		coreDNSport = 53
		option = "prefer_udp"
	}
	cm := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
		Data: map[string]string{
			"Corefile": fmt.Sprintf(coreDNSConf, coreDNSport, upstreamClusterName, coreDNSport, upsreamCoreDNSLB, option),
		},
	}
	_, err := k8s.CreateIgnoreExists(ctx, cli, cm)
	if err != nil {
		return err
	}

	labels := map[string]string{
		"app":  "cdc-dns-forward",
		"name": name,
	}

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
		Spec: corev1.ServiceSpec{
			Selector: labels,
			Type:     corev1.ServiceTypeClusterIP,
			Ports: []corev1.ServicePort{
				{
					Name:     "dns-udp",
					Protocol: corev1.ProtocolUDP,
					TargetPort: intstr.IntOrString{
						IntVal: coreDNSport,
						Type:   intstr.Int,
					},
					Port: 53,
				},
				{
					Name:     "metrics",
					Protocol: corev1.ProtocolTCP,
					Port:     9153,
					TargetPort: intstr.IntOrString{
						IntVal: 9153,
						Type:   intstr.Int,
					},
				},
			},
		},
	}
	if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeGKE {
		svc.Spec.Ports = append(svc.Spec.Ports, corev1.ServicePort{
			Name:     "dns-tcp",
			Protocol: corev1.ProtocolTCP,
			TargetPort: intstr.IntOrString{
				IntVal: coreDNSport,
				Type:   intstr.Int,
			},
			Port: 53,
		})
	}

	_, err = k8s.CreateIgnoreExists(ctx, cli, svc)
	if err != nil {
		return err
	}

	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "kube-system",
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: pointer.Int32(2),
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},

			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labels,
					Annotations: map[string]string{
						"scheduler.alpha.kubernetes.io/critical-pod": "",
					},
				},
			},
		},
	}

	_, err = controllerutil.CreateOrUpdate(ctx, cli, deploy, func() error {
		deploy.Spec.Replicas = pointer.Int32(2)
		deploy.Spec.Template.Spec = corev1.PodSpec{
			ServiceAccountName: "coredns",
			RestartPolicy:      corev1.RestartPolicyAlways,
			PriorityClassName:  "system-cluster-critical",
			Tolerations: []corev1.Toleration{
				{
					Key:      "CriticalAddonsOnly",
					Operator: corev1.TolerationOpExists,
				},
			},
			Volumes: []corev1.Volume{
				{
					Name: "config-volume",
					VolumeSource: corev1.VolumeSource{
						ConfigMap: &corev1.ConfigMapVolumeSource{
							DefaultMode: pointer.Int32(420),
							Items: []corev1.KeyToPath{
								{
									Key:  "Corefile",
									Path: "Corefile",
								},
							},
							LocalObjectReference: corev1.LocalObjectReference{
								Name: cm.Name,
							},
						},
					},
				},
			},
		}
		container := corev1.Container{
			Name:            "coredns",
			Image:           "gcr.io/pingcap-public/coredns:1.8.3",
			ImagePullPolicy: corev1.PullIfNotPresent,
			Args: []string{
				"-conf",
				"/etc/coredns/Corefile",
			},

			Resources: corev1.ResourceRequirements{
				Limits: corev1.ResourceList{
					corev1.ResourceCPU:    resource.MustParse("200m"),
					corev1.ResourceMemory: resource.MustParse("256Mi"),
				},
				Requests: corev1.ResourceList{
					corev1.ResourceCPU:    resource.MustParse("50m"),
					corev1.ResourceMemory: resource.MustParse("100Mi"),
				},
			},
			LivenessProbe: &corev1.Probe{
				TimeoutSeconds:      5,
				SuccessThreshold:    1,
				PeriodSeconds:       10,
				InitialDelaySeconds: 60,
				FailureThreshold:    5,
				ProbeHandler: corev1.ProbeHandler{
					HTTPGet: &corev1.HTTPGetAction{
						Path: "/health",
						Port: intstr.IntOrString{
							IntVal: 8080,
							Type:   intstr.Int,
						},
						Scheme: "HTTP",
					},
				},
			},
			ReadinessProbe: &corev1.Probe{
				TimeoutSeconds:   1,
				SuccessThreshold: 1,
				PeriodSeconds:    10,
				FailureThreshold: 3,
				ProbeHandler: corev1.ProbeHandler{
					HTTPGet: &corev1.HTTPGetAction{
						Path: "/ready",
						Port: intstr.IntOrString{
							IntVal: 8181,
							Type:   intstr.Int,
						},
						Scheme: "HTTP",
					},
				},
			},
			Ports: []corev1.ContainerPort{
				{
					Name:          "dns-udp",
					Protocol:      corev1.ProtocolUDP,
					ContainerPort: 8053,
				},
				{
					Name:          "dns-tcp",
					Protocol:      corev1.ProtocolTCP,
					ContainerPort: 8053,
				},
				{
					Name:          "metrics",
					Protocol:      corev1.ProtocolTCP,
					ContainerPort: 9153,
				},
			},
			VolumeMounts: []corev1.VolumeMount{
				{
					Name:      "config-volume",
					ReadOnly:  true,
					MountPath: "/etc/coredns",
				},
			},
		}
		deploy.Spec.Template.Spec.Containers = append(deploy.Spec.Template.Spec.Containers, container)
		return nil
	})
	return err
}

func ensureCoreDNSReady(ctx context.Context, cli client.Client, name string) (string, error) {
	app := new(appsv1.Deployment)
	err := cli.Get(ctx, types.NamespacedName{
		Namespace: "kube-system",
		Name:      name,
	}, app)
	if err != nil {
		return "", err
	}
	// deployment is ready
	if utils.IsStatusConditionTrue(app.Status.Conditions, appsv1.DeploymentAvailable) {
		svc := new(corev1.Service)
		err := cli.Get(ctx, types.NamespacedName{
			Namespace: "kube-system",
			Name:      name,
		}, svc)
		if err != nil {
			return "", err
		}
		if svc.Spec.ClusterIP == "" {
			return "", fmt.Errorf("wait coredns cluster IP allocated")
		}

		return svc.Spec.ClusterIP, nil
	}
	return "", fmt.Errorf("wait coredns ready")
}

func corednsName(cdc *cloudv1.TiCDC, destCluster *cloudv1.Cluster) string {
	if destCluster != nil {
		return fmt.Sprintf("coredns-%s", destCluster.Name)
	}
	return fmt.Sprintf("coredns-%s", cdc.Spec.UpstreamClusterName)
}

func ConvertCertV1alpha2ToV1(old *v1alpha2.Certificate) *certmanagerv1.Certificate {
	n := new(certmanagerv1.Certificate)
	n.APIVersion = certmanagerv1.SchemeGroupVersion.String()
	n.ObjectMeta = old.DeepCopy().ObjectMeta
	n.Spec = certmanagerv1.CertificateSpec{
		SecretName:  old.Spec.SecretName,
		Duration:    old.Spec.Duration,
		RenewBefore: old.Spec.RenewBefore,
		CommonName:  old.Spec.CommonName,
		Subject: &certmanagerv1.X509Subject{
			Organizations: old.Spec.Organization,
		},
		Usages:      ConvertKeyUsageV1alpha2ToV1(old.Spec.Usages),
		DNSNames:    old.Spec.DNSNames,
		IPAddresses: old.Spec.IPAddresses,
		IssuerRef:   old.Spec.IssuerRef,
	}
	return n
}

func ConvertKeyUsageV1alpha2ToV1(usages []v1alpha2.KeyUsage) []certmanagerv1.KeyUsage {
	var newUsages []certmanagerv1.KeyUsage

	for _, usage := range usages {
		newUsages = append(newUsages, certmanagerv1.KeyUsage(usage))
	}

	return newUsages
}

func (dc *ticdcCluster) CompatibleMultipleVersionsCertManager(defaultCert *v1alpha2.Certificate) client.Object {
	var obj client.Object = defaultCert
	if dc.cdc.RequireV1CertManager() {
		obj = ConvertCertV1alpha2ToV1(defaultCert)
	}
	return obj
}

// supportClusterID returns whether cdc support cluster-id parameter
func supportClusterID(version string) bool {
	if version == "" {
		return false
	}
	v := semver.New(removeVAndHash(version))
	return !v.LessThan(*semver.New("6.2.0")) || (v.Major == 6 && v.Minor == 2 && v.Patch == 0)
}

func removeVAndHash(v string) string {
	if v == "" {
		return v
	}
	v = versionHash.ReplaceAllLiteralString(v, "")
	v = strings.TrimSuffix(v, "-dirty")
	return strings.TrimPrefix(v, "v")
}

// getCrossRegionTiDBClusterName returns the label value of `cluster.cloud.pingcap.com/name`
// node scaler use this value to find the related `destCluster`
// if there is no `destCluster` returns empty string
func getCrossRegionTiDBClusterName(destCluster *cloudv1.Cluster) string {
	if destCluster != nil {
		return destCluster.Name
	}
	return ""
}

// getDestNamespace returns the namespace ticdc cluster deployed to
func getDestNamespace(
	upstreamCluster *cloudv1.Cluster,
	destCluster *cloudv1.Cluster) string {
	if destCluster != nil {
		return cloudv1.ClusterPrefixId(destCluster.Name)
	}
	// deploy in downstream and no downstream cluster, we should create a new namespace
	return cloudv1.ClusterPrefixId(upstreamCluster.Name) + "ticdc"
}

// getReferenceOwner returns the reference owner for TiCDC
// if `destCluster` is nil, use upstream cluster
func getReferenceOwner(
	upstreamCluster *cloudv1.Cluster,
	destCluster *cloudv1.Cluster) *cloudv1.Cluster {
	if destCluster != nil {
		return destCluster
	}
	return upstreamCluster
}

// ticdcNdeGroupNamePrefix returns the node group name prefix
// or node selector value of ticdc node
func ticdcNdeGroupNamePrefix(
	cdc *cloudv1.TiCDC,
	destCluster *cloudv1.Cluster) string {
	prefix := utils.TiCDCNodeGroupNamePrefix(isDRTiCDC(cdc), cdc.Spec.UpstreamClusterName, destCluster)
	return nodegroup.NodeGroupNamePrefix(prefix, string(cloudv1.ComponentTiCDC))
}