package cluster

import (
	"context"
	"fmt"
	"reflect"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	configsdk "github.com/tidbcloud/config-service/sdk"
	"github.com/tidbcloud/config-service/sdk/api"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/multierr"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/retry"
	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/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	reconcileerr "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/controllers/provider/gardener/version"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/audit"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gardener"
	gardencorev1beta1 "github.com/tidbcloud/infra-api/pkg/gardener/core/apis/v1beta1"
	"github.com/tidbcloud/infra-api/pkg/gitops"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
	"github.com/tidbcloud/infra-api/pkg/metrics"
	metricsctrl "github.com/tidbcloud/infra-api/pkg/metrics/controller"
	"github.com/tidbcloud/infra-api/pkg/nodegroup"
	"github.com/tidbcloud/infra-api/pkg/o11y"
	"github.com/tidbcloud/infra-api/pkg/pingcap/label"
	"github.com/tidbcloud/infra-api/pkg/progress"
	"github.com/tidbcloud/infra-api/pkg/providers/aws"
	"github.com/tidbcloud/infra-api/pkg/providers/gcp"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
)

const (
	Finalizer = "cluster.cloud.pingcap.com/finalizer"

	defaultWaitInterval = 10 * time.Second
)

var (
	clusterSetupWithManagerOnce sync.Once

	tracer = otel.Tracer("ClusterReconciler")
)

// ClusterReconciler reconciles a Cluster object
type ClusterReconciler struct {
	client.Client
	Log    logr.Logger
	Scheme *runtime.Scheme

	ProviderType string

	Gardener gardener.GardenerAPI
	AWS      aws.Client
	gcpApi   gcp.GCPAPI

	ConfigService  configsdk.Interface
	recorder       record.EventRecorder
	EnableGitOpsV2 bool
	Gitops         gitopsv2.Interface

	publicHostedZoneID     string
	publicHostedZoneDomain string

	// Cluster metric agent dependencies introduced by new o11y infra.
	o11yNewAgentEnabled     bool
	manifestDirectoryPrefix string
	o11yConfig              *cloudv1.ObservabilityInfraConfig
	o11yApi                 o11y.Api

	PDMixedDeployConfig *cloudv1.MixedDeployPDWithTiKV
	ExpFeatureGates     []cloudv1.ExpFeatureGate
}

func NewReconciler(
	mgr ctrl.Manager,
	g *gardener.GardEnvironment,
	providerType string,
	awsClient aws.Client,
	gcpApi gcp.GCPAPI,
	configService configsdk.Interface,
	cfg *cloudv1.InfraAPIConfig,
	o11yNewAgentEnabled bool,
	manifestDirectoryPrefix string,
	o11yConfig *cloudv1.ObservabilityInfraConfig,
	o11yApi o11y.Api,
) *ClusterReconciler {
	return &ClusterReconciler{
		Client:                  mgr.GetClient(),
		Log:                     ctrl.Log.WithName("controllers").WithName("Cluster").WithName(providerType),
		Scheme:                  mgr.GetScheme(),
		ProviderType:            providerType,
		Gardener:                g,
		AWS:                     awsClient,
		gcpApi:                  gcpApi,
		ConfigService:           configService,
		recorder:                mgr.GetEventRecorderFor("cluster-controller"),
		publicHostedZoneID:      cfg.DNS.PublicHostedZoneID,
		publicHostedZoneDomain:  cfg.DNS.PublicHostedZoneDomain,
		o11yNewAgentEnabled:     o11yNewAgentEnabled,
		manifestDirectoryPrefix: manifestDirectoryPrefix,
		o11yConfig:              o11yConfig,
		o11yApi:                 o11yApi,
		PDMixedDeployConfig:     &cfg.MixedDeployPDWithTiKV,
		ExpFeatureGates:         cfg.ExpFeatureGates,
	}
}

func (r *ClusterReconciler) SetUpGitopsV2(enabled bool) *ClusterReconciler {
	r.EnableGitOpsV2 = enabled
	if r.EnableGitOpsV2 {
		r.Gitops = &gitopsv2.Gitops{
			Cli: r.Client,
		}
	}
	return r
}

// reconcileContext is the local context of a single cluster reconciliation, it is NOT concurrency safe
type reconcileContext struct {
	*ClusterReconciler
	ClusterContext

	ctx context.Context

	stageBeginAt *time.Time

	providerGardenerSpec *cloudv1.GardenerProviderSpec
	networkGardenStatus  *cloudv1.GardenerNetworkStatus
	shootClient          client.Client
	shoot                *gardencorev1beta1.Shoot
	shootKubeConfig      []byte

	// transient state during a single reconciliation
	tc *v1alpha1.TidbCluster

	Log logr.Logger
}

// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters/finalizers,verbs=update
func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, err error) {
	start := time.Now()
	ctx, span := tracer.Start(ctx, "Cluster Reconcile")
	logger := log.FromContext(ctx)

	defer func() {
		tracerutil.EndSpan(span, err)
		metricsctrl.ReconcileDuration.WithLabelValues("cluster").Observe(time.Since(start).Seconds())

		params := []interface{}{"take", time.Since(start)}
		if err == nil {
			params = append(params, "result", result)
		}
		logger.Info("finish reconcile", params...)
	}()

	logger.Info("start reconcile cluster")

	c := &cloudv1.Cluster{}
	err = r.Get(ctx, req.NamespacedName, c)
	if err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	span.SetAttributes(
		attribute.String("cluster_name", req.Name),
		attribute.String("cluster_phase", string(c.Status.Phase)),
		attribute.String("cluster_namespace", req.Namespace),
		attribute.String("network_name", c.Spec.NetworkName),
		attribute.String("cluster_id", c.Labels[consts.ClusterTagKey]),
		attribute.String("project_id", c.Labels[consts.ProjectTagKey]),
		attribute.String("tenant_id", c.Labels[consts.TenantTagKey]),
		attribute.String("provider", c.Labels[cloudv1.ProviderTypeLabel]),
	)

	// build the reconcile context
	span.AddEvent("build the reconcile context")
	configProvider := gitops.NewHardcode()
	rx := &reconcileContext{ClusterReconciler: r, ctx: ctx}
	rx.Log = logger
	rx.Target = c
	rx.ConfigProvider = configProvider
	rx.GitOps, rx.Cluster = rx, rx
	if c.DeletionTimestamp != nil &&
		c.Status.Phase != cloudv1.ClusterPhaseDeleting && c.Status.Phase != cloudv1.ClusterPhasePendingDelete {
		// shortcut to PendingDelete phase whenever cluster is get deleted
		if err := rx.updateProgress(cloudv1.ClusterPhasePendingDelete, "cluster is pending to be deleted"); err != nil {
			return ctrl.Result{}, err
		}
	}
	if c.Status.Phase == "" {
		// shortcut to Pending phase when cluster is newly created
		if err := rx.updateProgress(cloudv1.ClusterPhasePending, "cluster is waiting to be reconciled"); err != nil {
			return ctrl.Result{}, err
		}
	}

	span.AddEvent("Get the cluster's network info")
	network := &cloudv1.Network{}
	err = r.Get(ctx, c.GetNetworkRef(), network)
	if err != nil {
		// existence of the referenced network should be ensured by webhook
		return ctrl.Result{}, err
	}

	span.AddEvent("Get the cluster's provider")
	p := &cloudv1.Provider{}
	err = r.Get(ctx, types.NamespacedName{Name: network.Spec.Provider}, p)
	if err != nil {
		return ctrl.Result{}, err
	}

	span.AddEvent("Get the cluster's network status")
	networkGardenStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
	if err != nil {
		return ctrl.Result{}, err
	}

	span.AddEvent("Get the cluster's shoot context")
	shootContext, err := r.Gardener.GetShootContext(ctx, networkGardenStatus.ShootName)
	if err != nil {
		// if shoot already hibernation, should be awake it
		if err == gardener.ErrShootAlreadyHibernation {
			span.AddEvent("Awake shoot")
			if err = r.Gardener.AwakeShoot(ctx, networkGardenStatus.ShootName); err != nil {
				return ctrl.Result{Requeue: true}, err
			}
			r.recorder.Event(network, corev1.EventTypeNormal, "AwakeShoot", fmt.Sprintf("Shoot %s awaked", networkGardenStatus.ShootName))

			// Sometimes update network will conflict, so we add a try for it
			_ = retry.RetryOnConflict(retry.DefaultRetry, func() error {
				network := &cloudv1.Network{}
				err = r.Get(ctx, c.GetNetworkRef(), network)
				if err != nil {
					return err
				}

				// make network reconcile, because we need recreate proxy svc and update status.
				if network.Annotations == nil {
					network.Annotations = make(map[string]string)
				}
				// Set reconcile timestamp to avoid hibernate the recently awakened shoot
				network.Annotations[consts.AnnOperationReconciled] = time.Now().String()

				return r.Update(ctx, network)
			})

			// alway requeue
			return ctrl.Result{Requeue: true}, err
		}
		if err == gardener.ErrShootNotReady {
			logger.Info("shoot not ready, requeueing...")
			span.AddEvent("shoot not ready, requeueing")
			return ctrl.Result{RequeueAfter: defaultWaitInterval}, nil
		}
		return ctrl.Result{}, err
	}

	shoot, err := r.Gardener.GetShoot(ctx, networkGardenStatus.ShootName)
	if err != nil {
		return ctrl.Result{}, err
	}
	rx.shoot = shoot

	span.AddEvent("Get forProvider spec")
	providerGardenerSpec, err := cloudv1.GardenerProviderSpecForProvider(p)
	if err != nil {
		return ctrl.Result{}, err
	}

	rx.Network = network
	rx.Provider = p
	rx.GitOps, rx.Cluster = rx, rx
	rx.networkGardenStatus = networkGardenStatus
	rx.shootClient = shootContext.Manager.GetClient()
	rx.shootKubeConfig = shootContext.Kubeconfig
	rx.providerGardenerSpec = providerGardenerSpec

	// ensure finalizer if the object is not yet deleted
	if c.DeletionTimestamp.IsZero() {
		if err := common.EnsureFinalizers(rx.ctx, rx.Client, rx.Target, Finalizer); err != nil {
			return ctrl.Result{}, err
		}
	}

	switch c.Status.Phase {
	case cloudv1.ClusterPhaseCreating, cloudv1.ClusterPhasePending:
		err = rx.create(ctx)
	case cloudv1.ClusterPhaseRunning, cloudv1.ClusterPhaseUpgrading,
		cloudv1.ClusterPhaseScaling, cloudv1.ClusterPhaseSuspend:
		err = rx.reconcile(ctx)
	case cloudv1.ClusterPhasePendingDelete:
		err = rx.pendingDelete(ctx)
	case cloudv1.ClusterPhaseDeleting:
		err = rx.delete(ctx)
	case cloudv1.ClusterPhaseUnavailable:
		// when cluster is unavailable, reconcile the suspend actions
		// for the TC only and observe until it recovers for safety
		errObserve := rx.observe(ctx)
		errTC := rx.reconcileTCSuspendActions(ctx)
		err = multierr.Combine(errTC, errObserve)
	default:
		// unknown phase, reconcile the suspend actions for the TC,
		// and try to recover the real phase via observation
		errObserve := rx.observe(ctx)
		errTC := rx.reconcileTCSuspendActions(ctx)
		err = multierr.Combine(errTC, errObserve)
	}

	return common.ReconcileErr(logger, err)
}

func (r *reconcileContext) ensureClusterStorageClass(ctx context.Context) error {
	k8sVersionLessThan118, err := version.CompareVersions(r.shoot.Spec.Kubernetes.Version, "<", "1.18")
	if err != nil {
		log.FromContext(ctx).Error(err, "failed to compare version", "version", r.shoot.Spec.Kubernetes.Version)
	}

	// ref https://github.com/tidbcloud/gardener-extension-provider-aws/blob/v.1.7.2-pingcap/docs/usage-as-end-user.md#csi-volume-provisioners
	// Every AWS shoot cluster that has at least Kubernetes v1.18 will be deployed with the AWS EBS CSI driver
	// Currently, there is only one shoot in prod still running 1.16.9, if we upgrade it finally we can remove this code.
	// You can using this query to check:
	// count by (kubelet_version,cluster) (kube_node_info{cluster=~"shoot.+", seed_provider="aws", kubelet_version="v1.16.9"})))
	if k8sVersionLessThan118 {
		err := cloudv1.EnsureClusterStorageClassAWSInTreeProvisioner(r.ctx, r.shootClient, r.Provider, r.Target)
		if err != nil {
			return err
		}
	} else {
		err := cloudv1.EnsureClusterStorageClass(r.ctx, r.shootClient, r.Provider, r.Target)
		if err != nil {
			return err
		}
	}

	return nil
}

func (r *reconcileContext) create(ctx context.Context) (errResult error) {
	var span trace.Span
	ctx, span = tracer.Start(ctx, "create cluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	if r.Network.Status.Phase == cloudv1.NetworkPhaseCreating {
		span.AddEvent("Waiting network ready")
		if err := r.updateProgress(cloudv1.ClusterPhasePending, "waiting for network to be created"); err != nil {
			return err
		}
		r.Log.V(4).Info("waiting network ready", "current phase", r.Network.Status.Phase)
		return &reconcileerr.RequeueAfterError{RequeueAfter: defaultWaitInterval}
	}

	if err := r.setDefaultDNSIdentity(); err != nil {
		return err
	}

	if err := r.ensureClusterStorageClass(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "StorageClass")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "creating kms key"); err != nil {
		return err
	}
	if err := r.ensureClusterNamespace(); err != nil {
		return err
	}
	if err := r.ensureEncryptionKey(ctx); err != nil {
		return err
	}
	if err := r.postProcessEncryptionKey(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "KMS")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb cluster"); err != nil {
		return err
	}
	if err := r.ensureTidbCluster(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "TiDBCluster")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb initializer"); err != nil {
		return err
	}
	if err := r.ensureTidbInitializer(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "TiDBInitializer")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring admin tidb"); err != nil {
		return err
	}
	if err := r.ensureAdminTiDB(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "AdminTiDB")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring tidb monitor"); err != nil {
		return err
	}
	if err := r.ensureTidbMonitor(ctx); err != nil {
		return err
	}
	if r.o11yNewAgentEnabled && utils.MatchRegion(r.Network.Spec.Provider, r.Network.Spec.Region, r.shoot.Name, r.o11yConfig.EffectiveRegion) {
		if err := r.ensureClusterAgent(ctx); err != nil {
			return err
		}
	}
	r.progress(ctx, r.Target, "O11Y")

	if err := r.reconcileBackupRBAC(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "BackupRBAC")

	if err := r.observeNodeProgressAndReason(); err != nil {
		// node progress does not block the reconciliation
		r.Log.Error(err, "error observing node progress and reason")
	}

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring endpoints"); err != nil {
		return err
	}
	errs := multierr.Combine(
		r.reconcilePublicEndpoint(ctx),
		r.reconcilePrivateEndpoint(ctx),
	)
	if errs != nil {
		return errs
	}
	r.progress(ctx, r.Target, "Endpoints")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring privatelink service"); err != nil {
		return err
	}

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "wait cluster ready"); err != nil {
		return err
	}
	if err := r.ensureReady(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "WaitClusterReady")

	if err := r.updateProgress(cloudv1.ClusterPhaseCreating, "ensuring SEM"); err != nil {
		return err
	}
	if err := r.ensureSecurityEnhancedMode(ctx); err != nil {
		return err
	}
	r.progress(ctx, r.Target, "SEM")

	r.Log.Info("mark cluster become running when creating cluster", "sinceCreate", time.Since(r.Target.CreationTimestamp.Time))
	r.progress(ctx, r.Target, "All")
	if err := r.markAsRunning(ctx); err != nil {
		return err
	}

	// creation complete, requeue and forward the cluster to reconcile() phase
	return &reconcileerr.RequeueAfterError{RequeueAfter: defaultWaitInterval, Caused: nil}
}

func (r *reconcileContext) reconcile(ctx context.Context) (errResult error) {
	var span trace.Span
	ctx, span = tracer.Start(ctx, "reconcile cluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	if err := r.setDefaultDNSIdentity(); err != nil {
		return err
	}

	errs := multierr.Combine(
		r.reconcilePublicEndpoint(ctx),
		r.reconcilePrivateEndpoint(ctx),
	)
	if err := r.reconcileTidbCluster(ctx); err != nil {
		errs = multierr.Append(errs, err)
	}

	var normalRequeues error
	plNormalRequeue, err := r.reconcilePrivateLinkService(ctx)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	normalRequeues = multierr.Append(normalRequeues, plNormalRequeue)

	if r.tc == nil { // r.tc get in preview reconcile**Endpoint now
		return errs
	}

	res, err := audit.Reconcile(ctx, r.Network, r.Target, r.Client, r.tc, r.shootClient, true /*advanceSTS*/)
	if err != nil {
		errs = multierr.Append(errs, err)
	}
	if !res.IsZero() {
		err := &reconcileerr.RequeueAfterError{RequeueAfter: res.RequeueAfter}
		errs = multierr.Append(errs, err)
	}

	// Enable o11y agent for existing tidb clusters.
	if r.o11yNewAgentEnabled && utils.MatchRegion(r.Network.Spec.Provider, r.Network.Spec.Region, r.shoot.Name, r.o11yConfig.EffectiveRegion) {
		if err := r.ensureClusterAgent(ctx); err != nil {
			errs = multierr.Append(errs, err)
		}
	}

	if errs != nil {
		return errs
	}

	// observe the latest status after an success reconciliation
	return multierr.Append(normalRequeues, r.observe(ctx))
}

func (r *reconcileContext) observe(ctx context.Context) (errResult error) {
	_, span := tracer.Start(ctx, "observe cluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	var errs error

	span.AddEvent("observeTidbClusterStatus")
	if err := r.observeTidbClusterStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("observeMetricsStatus")
	if err := r.observeMetricsStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("observeProxyStatus")
	if err := r.observeProxyStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("observeObjectStoreStatus")
	if err := r.observeObjectStoreStatus(); err != nil {
		errs = multierr.Append(errs, err)
	}

	if errs != nil {
		return errs
	}

	// schedule an observation after to keep the status up to date
	// TODO(aylei): deploy a local controller to shoot to report tc status
	return &reconcileerr.RequeueAfterError{RequeueAfter: observeAfter}
}

func (r *reconcileContext) observeTidbClusterStatus() error {
	if err := r.refreshTidbClusterStatus(); err != nil {
		return err
	}
	resolver := &TiDBClusterStatusResolver{
		Client:              r.shootClient,
		Machine:             r.Provider,
		Cluster:             r,
		Target:              r.Target,
		TiDBCluster:         r.tc,
		Log:                 r.Log,
		PDMixedDeployConfig: r.PDMixedDeployConfig,
	}
	return r.ensureStatus(func() error {
		if err := resolver.Resolve(r.ctx); err != nil {
			return err
		}

		switch r.ProviderType {
		case consts.ProviderTypeGCP:
			if err := cloudv1.MutateGcpClusterStatus(r.Target, func(status *cloudv1.GcpClusterStatus) error {
				if status.Kubernetes == nil {
					status.Kubernetes = &cloudv1.KubernetesStatus{}
				}
				status.Kubernetes.Namespace = r.tc.Namespace
				return nil
			}); err != nil {
				return err
			}
		case consts.ProviderTypeAWS:
			if err := cloudv1.MutateAwsClusterStatus(r.Target, func(status *cloudv1.AwsClusterStatus) error {
				if status.Kubernetes == nil {
					status.Kubernetes = &cloudv1.KubernetesStatus{}
				}
				status.Kubernetes.Namespace = r.tc.Namespace
				return nil
			}); err != nil {
				return err
			}
		default:
			return fmt.Errorf("unsupported provider type %s", r.ProviderType)
		}
		return nil
	})
}

func (r *reconcileContext) pendingDelete(ctx context.Context) (errResult error) {
	_, span := tracer.Start(ctx, "pendingDelete cluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	span.AddEvent("Delete all uncompleted backups")
	if names, err := r.getNotFinishedBackupNames(); err != nil {
		return err
	} else if len(names) > 0 {
		if err := r.deleteNotFinishedBackupNames(names); err != nil {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       err,
			}
		} else {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       errors.Errorf("wait for backups %v to be removed", names),
			}
		}
	}

	span.AddEvent("Delete all restores")
	if names, err := r.getRestoreNames(); err != nil {
		return err
	} else if len(names) > 0 {
		if err := r.deleteRestores(names); err != nil {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       err,
			}
		} else {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       errors.Errorf("wait for restores %v to be removed", names),
			}
		}
	}

	// check all imports are deleted
	span.AddEvent("Delete all imports")
	if names, err := r.getImportNames(); err != nil {
		return err
	} else if len(names) > 0 {
		if err := r.deleteImports(names); err != nil {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       err,
			}
		} else {
			return &reconcileerr.RequeueAfterError{
				RequeueAfter: defaultWaitInterval,
				Caused:       errors.Errorf("wait for imports %v to be removed", names),
			}
		}
	}

	if err := r.updateProgress(cloudv1.ClusterPhaseDeleting, "cluster is being deleted"); err != nil {
		return err
	}

	// pendingDelete complete, forward to next phase
	return &reconcileerr.RequeueAfterError{RequeueAfter: defaultWaitInterval}
}

func (r *reconcileContext) deleteNotFinishedBackupNames(names []string) error {
	for _, name := range names {
		if err := r.Client.Delete(r.ctx, &cloudv1.Backup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      name,
				Namespace: r.Target.Namespace,
			},
		}); client.IgnoreNotFound(err) != nil {
			return err
		}
	}
	return nil
}

func (r *reconcileContext) getNotFinishedBackupNames() ([]string, error) {
	var backupList cloudv1.BackupList
	if err := r.Client.List(r.ctx, &backupList, client.InNamespace(r.Target.Namespace), client.MatchingFields(map[string]string{
		"spec.clusterName": r.Target.Name,
	})); err != nil {
		return nil, err
	}
	var names []string
	for _, item := range backupList.Items {
		if item.Status.Phase != cloudv1.BackupComplete && item.Status.Phase != cloudv1.BackupFailed {
			names = append(names, item.Name)
		}
	}
	return names, nil
}

func (r *reconcileContext) deleteRestores(names []string) error {
	for _, name := range names {
		if err := r.Client.Delete(r.ctx, &cloudv1.Restore{
			ObjectMeta: metav1.ObjectMeta{
				Name:      name,
				Namespace: r.Target.Namespace,
			},
		}); client.IgnoreNotFound(err) != nil {
			return err
		}
	}
	return nil
}

func (r *reconcileContext) getRestoreNames() ([]string, error) {
	var restoreList cloudv1.RestoreList
	if err := r.Client.List(r.ctx, &restoreList, client.InNamespace(r.Target.Namespace), client.MatchingFields(map[string]string{
		"spec.clusterName": r.Target.Name,
	})); err != nil {
		return nil, err
	}
	var names []string
	for _, item := range restoreList.Items {
		names = append(names, item.Name)
	}
	return names, nil
}

func (r *reconcileContext) deleteImports(names []string) error {
	for _, name := range names {
		if err := r.Client.Delete(r.ctx, &cloudv1.Import{
			ObjectMeta: metav1.ObjectMeta{
				Name:      name,
				Namespace: r.Target.Namespace,
			},
		}); client.IgnoreNotFound(err) != nil {
			return err
		}
	}
	return nil
}

func (r *reconcileContext) getImportNames() ([]string, error) {
	var importList cloudv1.ImportList
	if err := r.Client.List(r.ctx, &importList, client.InNamespace(r.Target.Namespace), client.MatchingFields(map[string]string{
		"spec.clusterName": r.Target.Name,
	})); err != nil {
		return nil, err
	}
	var names []string
	for _, item := range importList.Items {
		names = append(names, item.Name)
	}
	return names, nil
}

func (r *reconcileContext) delete(ctx context.Context) (errResult error) {
	_, span := tracer.Start(ctx, "delete cluster")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	var errs error
	span.AddEvent("deleteInitializer")
	errs = multierr.Append(errs, r.deleteInitializer())

	span.AddEvent("deleteEndpoints")
	if err := r.deleteEndpoints(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("deletePrivateLinkService")
	if err := r.deletePrivateLinkService(ctx); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("deleteTidbCluster")
	if err := r.deleteTidbCluster(); err != nil {
		errs = multierr.Append(errs, err)
	}

	if r.o11yNewAgentEnabled && utils.MatchRegion(r.Network.Spec.Provider, r.Network.Spec.Region, r.shoot.Name, r.o11yConfig.EffectiveRegion) {
		span.AddEvent("deleteClusterAgent")
		if err := r.deleteClusterAgent(ctx); err != nil {
			errs = multierr.Append(errs, err)
		}
	}

	span.AddEvent("deleteEncryptionKey")
	if err := r.deleteEncryptionKey(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("reclaimPVs")
	if err := r.reclaimPVs(); err != nil {
		errs = multierr.Append(errs, err)
	}

	span.AddEvent("deleteNodeGroups")
	if err := r.deleteNodeGroups(); err != nil {
		errs = multierr.Append(errs, err)
	}
	if errs != nil {
		return errs
	}

	span.AddEvent("RemoveFinalizer")
	controllerutil.RemoveFinalizer(r.Target, Finalizer)
	if err := r.Update(r.ctx, r.Target); err != nil {
		return err
	}
	r.Log.Info("cleanup complete, delete cluster")
	return nil
}

func (r *reconcileContext) progress(ctx context.Context, cluster *cloudv1.Cluster, name string) {
	if !(cluster.Status.Phase == cloudv1.ClusterPhaseCreating || cluster.Status.Phase == cloudv1.ClusterPhasePending) {
		return
	}

	updated := progress.UpdateAnnForFirstPassIfNeed(cluster, name)
	if !updated {
		return
	}

	r.Log.Info("pass progress when creating cluster", "reason", name, "sinceCreate", time.Since(r.Target.CreationTimestamp.Time))
	err := r.Client.Update(ctx, cluster)
	if err != nil {
		r.Log.Error(err, "failed to update cluster")
	}
}

func (r *reconcileContext) updateProgress(phase cloudv1.ClusterPhase, reason string) error {
	if phase == cloudv1.ClusterPhaseCreating || phase == cloudv1.ClusterPhasePending {
		if r.stageBeginAt != nil {
			metrics.ClusterCreationDuration.WithLabelValues(r.Target.Status.Reason).Observe(time.Since(*r.stageBeginAt).Seconds())
		}
	}
	now := time.Now()
	r.stageBeginAt = &now
	if err := r.ensureStatus(func() error {
		r.Target.Status.Phase = phase
		r.Target.Status.Reason = reason
		return nil
	}); err != nil {
		return errors.Annotate(err, "update progress")
	}
	return nil
}

// ensureStatus update the cluster status if necessary
func (r *reconcileContext) ensureStatus(mutateFn func() error) error {
	beforeMutation := r.Target.DeepCopy()
	if err := mutateFn(); err != nil {
		return err
	}
	if reflect.DeepEqual(beforeMutation, r.Target) {
		// no changed, skip updating
		// Note: the local status may not be same as the truth, but we will reach eventually-consistency via reconciliation
		return nil
	}

	return r.Status().Update(r.ctx, r.Target)
}

// ensureAnnotation update the cluster annotation if necessary
func (r *reconcileContext) ensureAnnotations(mutateFn func() error) error {
	beforeMutation := r.Target.DeepCopy().Annotations
	if err := mutateFn(); err != nil {
		return err
	}
	if reflect.DeepEqual(beforeMutation, r.Target.Annotations) {
		// no changed, skip updating
		return nil
	}

	return r.Update(r.ctx, r.Target)
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager, pdt predicate.Predicate, maxConcurrentReconciles int) error {
	p, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			cloudv1.ProviderTypeLabel: r.ProviderType,
		},
	})
	if err != nil {
		return err
	}

	// SetupWithManager will be called for different cluster provider multi time, but we only need to add index filed once
	clusterSetupWithManagerOnce.Do(func() {
		if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Backup{}, "spec.clusterName", func(obj client.Object) []string {
			backup := obj.(*cloudv1.Backup)
			return []string{backup.Spec.ClusterName}
		}); err != nil {
			return
		}

		if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Restore{}, "spec.clusterName", func(obj client.Object) []string {
			restore := obj.(*cloudv1.Restore)
			return []string{restore.Spec.ClusterName}
		}); err != nil {
			return
		}

		if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.TiCDC{}, "spec.clusterName", func(obj client.Object) []string {
			ticdc := obj.(*cloudv1.TiCDC)
			return []string{ticdc.Spec.ClusterName}
		}); err != nil {
			return
		}

		if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Import{}, "spec.clusterName", func(obj client.Object) []string {
			importObj := obj.(*cloudv1.Import)
			return []string{importObj.Spec.ClusterName}
		}); err != nil {
			return
		}
		// add index to cluster object, used for list clusters in network
		if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Cluster{}, "spec.networkName", func(obj client.Object) []string {
			clusterObj := obj.(*cloudv1.Cluster)
			return []string{clusterObj.Spec.NetworkName}
		}); err != nil {
			return
		}
	})
	if err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Cluster{}, builder.WithPredicates(pdt)).
		WithOptions(controller.Options{RateLimiter: common.DefaultRateLimiter(), MaxConcurrentReconciles: maxConcurrentReconciles}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		WithEventFilter(p).
		WithEventFilter(predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{})).
		Complete(r)
}

func (r *reconcileContext) GetClusterConfig(_ context.Context, _ *cloudv1.Cluster) (*api.DescribeClusterConfigOutput, error) {
	return r.getClusterConfig()
}

func (r *reconcileContext) GetComponentNodeLabels(componentType cloudv1.ComponentType) map[string]string {
	return map[string]string{nodegroup.GardenerWorkerPrefixLabelName: nodegroup.NodeGroupNamePrefix(r.Target.PrefixedId(), string(componentType))}
}

func (r *reconcileContext) GetComponentPodLabels(componentType cloudv1.ComponentType) map[string]string {
	var tcName string
	if r.tc != nil {
		tcName = r.tc.Name
	}
	return map[string]string{
		label.ComponentLabelKey: string(componentType),
		label.InstanceLabelKey:  tcName,
	}
}

func (r *reconcileContext) GetNodeZoneFromLabels(labels map[string]string) string {
	if labels == nil {
		return ""
	}
	return labels[nodegroup.GardenerNodeAvailableZoneInfo]
}

func (r *reconcileContext) GetNodeInstanceTypeFromLabels(labels map[string]string) string {
	if labels == nil {
		return ""
	}
	return labels[corev1.LabelInstanceTypeStable]
}

func (r *reconcileContext) Overlay(tc *v1alpha1.TidbCluster) error {
	for _, f := range [](func(*v1alpha1.TidbCluster) error){
		r.OverlayTiFlash,
		r.overlayTags,
		r.overlayAnnotations,
		r.overlayReplicas,
		r.overlayTiDBEndpoint,
		r.overlayTiKVKmsCredentials,
		r.overlayBillingLabels,
		r.overlaySuspendActions,
	} {
		if err := f(tc); err != nil {
			return err
		}
	}
	return nil
}