package cluster

import (
	"context"
	"crypto/tls"
	"fmt"
	"net/http"
	"time"

	"cloud.google.com/go/pubsub"
	"github.com/go-logr/logr"
	"github.com/prometheus/client_golang/prometheus"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/ptr"
	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/log"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/eks-provider/controllers"
	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/utils/configmanager"
	"github.com/tidbcloud/eks-provider/controllers/utils/maxtimes"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/account"
	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/gitops"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/metrics"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/o11y/tracing"
	"github.com/tidbcloud/infra-provider/pkg/taskevents"
	tracerutil "github.com/tidbcloud/infra-provider/pkg/tracer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

var tracer = otel.Tracer("ClusterReconciler")

// ClusterReconciler reconciles a Cluster object
type ClusterReconciler struct {
	Logger logr.Logger

	EventRecorder record.EventRecorder
	client.Client
	LegacyManager         *k8s.Manager
	CachedProvider        dataplane.EnvProviderManager
	ConfigManager         configmanager.Interface
	CloudConfigProvider   config.CloudConfigProvider
	Cache                 client.Reader
	Config                config.RootConfig
	AcctInfoGetter        account.Getter
	DataplaneHTTPEndpoint DataplaneHTTPEndpoint // inject DataplaneHTTPEndpoint for integration test
	PDAuthFunc            configmanager.PDAuth

	AWSCli     aws.Client
	GCPCli     gcp.Client
	TopicCli   *pubsub.Topic
	httpClient *http.Client

	MaxTimes maxtimes.MaxTimes
}

//+kubebuilder:rbac:groups=cloud.pingcap.com.infra.pingcap.com,resources=clusters,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.pingcap.com.infra.pingcap.com,resources=clusters/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com.infra.pingcap.com,resources=clusters/finalizers,verbs=update
//+kubebuilder:rbac:groups=container.gcp.crossplane.io,resources=nodepools,verbs=get;list;watch;create;update;patch;delete

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)

	logger.Info("Reconciling Cluster")
	start := time.Now()
	var cluster cloudv1.Cluster
	defer func() {
		metricsctrl.ReconcileDuration.
			WithLabelValues("cluster").Observe(time.Since(start).Seconds())
		logger.Info("Finish reconcile", "take", time.Since(start), "phase", cluster.Status.Phase)
	}()
	if err := r.Cache.Get(ctx, req.NamespacedName, &cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get cluster %s: %v", req.NamespacedName, err)
	}

	clusterVersion := ""
	if cluster.Status.DB != nil {
		clusterVersion = cluster.Status.DB.Version
	}
	clusterPaused := "false"
	if ptr.Deref(cluster.Spec.Paused, false) {
		clusterPaused = "true"
	}
	clusterDryrunMetricsLabels := map[string]string{
		"provider":   cluster.GetProviderType(),
		"region":     cluster.GetRegion(),
		"name":       cluster.Name,
		"namespace":  cluster.Namespace,
		"tenant_id":  cluster.GetTenantID(),
		"project_id": cluster.GetProjectID(),
		"version":    clusterVersion,
		"status":     string(cluster.Status.Phase),
		"paused":     clusterPaused,
		"result":     "",
	}

	assumedManageIAMRoleName := r.Config.FullManage.AssumeRoleName
	if r.Config.DryRun.Enabled {
		// When the cluster-controller runs in dryrun/readonly mode, we should use another IAM role
		// `readonly-{env}-{region}-cluster-controller-assumed-role` to assume. Hence, we pass this role to `NetworkKubeCluster`
		// to build dataplane k8s client.
		// Info:
		//   1. the readonly assumed AWS IAM role (in general, it's `readonly-{env}-{region}-cluster-controller-assumed-role`) is granted with
		//      readonly permissions by `kube-system/aws-auth` in rbac.go.
		assumedManageIAMRoleName = r.Config.FullManage.ReadonlyAssumeRoleName

		if !r.MaxTimes.Do(req.NamespacedName.String()) {
			logger.Info("ignore this key, max times has been exceeded")
			return ctrl.Result{}, nil
		}

		if !cluster.FullSynced() || !cluster.DeletionTimestamp.IsZero() || ptr.Deref(cluster.Spec.Paused, false) {
			logger.Info("Skipping unsynced/deleting/paused cluster in dryrun mode", "cluster", req.NamespacedName)

			r.MaxTimes.Done(req.NamespacedName.String())

			clusterDryrunMetricsLabels["result"] = "skipped"
			metrics.ClusterDryrun.With(clusterDryrunMetricsLabels).Set(1)
			return ctrl.Result{}, nil
		}
	}

	if cluster.Spec.Paused != nil && *cluster.Spec.Paused {
		logger.Info("Skipping paused cluster", "cluster", req.NamespacedName)
		return ctrl.Result{}, nil
	}

	// The ConfigRevisionRef.name should always have been set by webhook defined in [config-manager](../../../config-manager/pkg/webhooks/cluster_webhook.go)
	// when the reconcile loop is triggered. If it is not set, there must be something wrong!
	if cluster.Spec.ConfigRevisionRef == nil || cluster.Spec.ConfigRevisionRef.Name == "" {
		metrics.ConfigRevisionRefIsNil.WithLabelValues(cluster.Name).Set(1)
		err := fmt.Errorf("configRevision not found, cluster=%s", req.NamespacedName)
		logger.Error(err, "cluster without revision, skip")
		r.EventRecorder.Event(&cluster, corev1.EventTypeWarning, "ReconcileFail", err.Error())
		// Reconcile loop will be trigger again after the ConfigRevisionRef.Name has been set, so just skip current loop here.
		return ctrl.Result{}, nil
	} else {
		metrics.ConfigRevisionRefIsNil.DeletePartialMatch(prometheus.Labels{"cluster_id": cluster.Name})
	}

	ep, err := r.CachedProvider.GetEnvProvider(ctx, &cluster)
	if err != nil {
		r.EventRecorder.Eventf(&cluster, corev1.EventTypeWarning, "CannotNewEnvProvider", "cannot new env provier: %v", err)

		if cluster.Spec.NetworkName == "" {
			// only return err for NEW cluster
			// legacy cluster will try to use legacy env provider
			return ctrl.Result{}, fmt.Errorf("cannot new env provider: %w", err)
		}
	}

	var legacyNetwork *cloudv1.Network
	var legacyAcct *cloudv1.Account
	var legacyRemote k8s.ClientGetter

	networkKey := cluster.GetNetworkRef()

	hasNetwork := false

	if networkKey.Name != "" {
		hasNetwork = true
		var network cloudv1.Network
		if err := r.Cache.Get(ctx, networkKey, &network); err != nil {
			if !apierrors.IsNotFound(err) {
				return ctrl.Result{}, fmt.Errorf("unable to get network %s: %v", cluster.Spec.NetworkName, err)
			}
			hasNetwork = false
		}

		if hasNetwork {
			providerType := network.Spec.Provider
			if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeGKE {
				logger.Info("ignore unknown provider type", "provider", providerType)
				return ctrl.Result{}, nil
			}
			if network.Status.Phase == cloudv1.NetworkPhaseNormal || network.Status.Phase == cloudv1.NetworkPhasePendingDelete {
				var assumedManagerRole *k8s.AssumeRole
				if r.Config.DryRun.Enabled && providerType == consts.ProviderTypeEKS {
					assumedManagerRole = &k8s.AssumeRole{
						// the arguments `assumedManageIAMRoleName` will be overridden to a readonly IAM role
						RoleARN:    util.GetRoleARN(network.Status.IAM.AccountID, assumedManageIAMRoleName),
						ExternalID: r.Config.FullManage.AssumeRoleExternalID,
					}
				}
				if err := r.LegacyManager.Register(ctx, &k8s.NetworkKubeCluster{Network: &network, AWS: assumedManagerRole}); err != nil {
					logger.Info("failed to register network", "network", networkKey)
					return ctrl.Result{}, err
				}
			}

			if network.Status.Phase == cloudv1.NetworkPhaseHibernating {
				logger.Info("Wait network status from hibernating to hibernated")
				return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
			}

			if network.Status.Phase == cloudv1.NetworkPhaseHibernated {
				network.Status.Phase = cloudv1.NetworkPhaseWakeUp
				if network.Annotations == nil {
					network.Annotations = make(map[string]string)
				}
				// record the time when the network is waken up
				network.Annotations[cloudv1.NetworkAwakeTimeAnnotation] = time.Now().Format(time.RFC3339)
				err := r.Client.Status().Update(context.TODO(), &network)
				if err != nil {
					return ctrl.Result{}, fmt.Errorf("failed update network status to waking up: %v", err)
				}
			}

			if network.Status.Phase == cloudv1.NetworkPhaseWakeUp {
				logger.Info("waiting network status from wakingup to normal", "network", cluster.Spec.NetworkName)
				return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
			}
			var acct cloudv1.Account
			if network.Spec.AccountRef != nil {
				if err := r.Cache.Get(ctx, types.NamespacedName{Name: network.Spec.AccountRef.Name}, &acct); err != nil {
					return ctrl.Result{}, fmt.Errorf("unable to get account %s: %v", network.Spec.AccountRef.Name, err)
				}
			}
			remote, err := r.LegacyManager.ClientGetter(ctx, networkKey.String())
			if err != nil { // we need to wait manager ready
				return ctrl.Result{}, fmt.Errorf("failed to set remote client: %v", err)
			}

			legacyNetwork = &network
			legacyAcct = &acct
			legacyRemote = remote
		}
	}

	if !hasNetwork {
		providerType := r.Config.Predicate.MatchLabels[cloudv1.ProviderTypeLabel]
		if providerType == consts.ProviderTypeEKS {
			var kubeCluster k8s.KubeCluster = k8s.ClusterKubeCluster{
				Cluster: &cluster,
				AWS: &k8s.AssumeRole{
					RoleARN:    util.GetRoleARN(ep.GetAccount(), assumedManageIAMRoleName),
					ExternalID: r.Config.FullManage.AssumeRoleExternalID,
				},
			}
			if err := r.LegacyManager.Register(ctx, kubeCluster); err != nil {
				logger.Info("failed to register dataplane controller", "kubernetes", kubeCluster.GetName())
				return ctrl.Result{}, err
			}
		} else if providerType == consts.ProviderTypeGKE && cluster.GetDeletionTimestamp().IsZero() {
			var kubeCluster k8s.KubeCluster = k8s.ClusterKubeCluster{
				Cluster: &cluster,
			}
			if err := r.LegacyManager.Register(ctx, kubeCluster); err != nil {
				logger.Info("failed to register dataplane controller", "kubernetes", kubeCluster.GetName())
				return ctrl.Result{}, err
			}
		}
	}

	c, err := (&task.Context{
		EnvProvider:    ep,
		Local:          r.Client,
		LegacyRemote:   legacyRemote,
		EventRecorder:  r.EventRecorder,
		Config:         r.Config,
		ConfigProvider: gitops.NewHardcode(),
		Cluster:        &cluster,
		Provider:       r.CloudConfigProvider,
		ConfigManager:  r.ConfigManager,
		AcctInfoGetter: r.AcctInfoGetter,
		Logger:         logger,

		LegacyNetwork: legacyNetwork,
		LegacyAccount: legacyAcct,
	}).Default(ctx)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to new context: %v", err)
	}

	remoteCli := c.EnvProvider.GetClient()
	dataplaneHTTPEndpoint := r.DataplaneHTTPEndpoint
	if dataplaneHTTPEndpoint == nil {
		gatewayEnabled := config.FeatureEnabled(r.Config.FeatureGates, config.UseProxyGatewayIngress, cluster.GetTenantID(), cluster.GetProjectID())
		if gatewayEnabled {
			dataplaneHTTPEndpoint = NewServiceHTTPEndpoint(remoteCli, types.NamespacedName{Namespace: "tidb-admin", Name: "clustergateway-kong-proxy"})
		} else {
			dataplaneHTTPEndpoint = NewServiceHTTPEndpoint(remoteCli, types.NamespacedName{Namespace: "tidb-admin", Name: "proxy-cluster-proxy"})
		}
	}

	c.PDClient, err = dataplaneHTTPEndpoint.GetPDClient(ctx, r.httpClient, r.PDAuthFunc, cluster.Name, r.Config.DryRun.Enabled)
	if err != nil {
		return ctrl.Result{}, err
	}

	r.migrateFeatureAnnotations(ctx, c)

	tasks := GetClusterTasks(r, c)

	var res ctrl.Result

	if cluster.DeletionTimestamp.IsZero() &&
		(c.Cluster.Status.Phase == "" || c.Cluster.Status.Phase == cloudv1.ClusterPhasePending || c.Cluster.Status.Phase == cloudv1.ClusterPhaseCreating) {
		hook := func(ctx context.Context, task deploy.Task) {
			name := task.Name()

			if c.Cluster.Annotations == nil {
				c.Cluster.Annotations = make(map[string]string)
			}
			age := time.Since(c.Cluster.CreationTimestamp.Time)
			logger.Info("pass task when creating cluster",
				"task", name, "sinceCreate", age)
			metrics.ClusterFirstPassStageSeconds.WithLabelValues(name).Observe(age.Seconds())

			if taskevents.IsEnabled(r.Config) {
				taskevents.PublishTaskEvent(ctx, r.TopicCli, taskevents.TaskEvent{
					Resource:    "cluster",
					Namespace:   c.Cluster.Namespace,
					Name:        c.Cluster.Name,
					Task:        name,
					Status:      "done",
					ResourceAge: age.Seconds(),
					EventTime:   time.Now(),
				})
			}
		}

		var span trace.Span
		ctx, span = tracer.Start(tracing.ExtractTraceContext(ctx, &cluster), "Cluster Reconcile")
		span.SetAttributes(
			attribute.String("tenant_id", cluster.GetTenantID()),
			attribute.String("project_id", cluster.GetProjectID()),
			attribute.String("namespace", cluster.Namespace),
			attribute.String("cluster_id", cluster.Name),
		)
		defer func() { tracerutil.EndSpan(span, err) }()

		for i, t := range tasks {
			tasks[i] = deploy.NewTaskWraper(t, hook, tracer)
		}
	}

	d := deploy.NewDeployer(tasks, nil, logger)
	if cluster.DeletionTimestamp.IsZero() {
		res, err = updateCreate(ctx, c, d.Deploy(ctx))
	} else {
		// When running deletion tasks, we need to initialize the TCGetter,
		// otherwise it will panic when some tasks try to use TCGetter.
		// Because we have already checked the nil value in implementation of TCGetter,
		// it's safe to use nil values to initialize it.
		if cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2) {
			c.TCGetter = &tidbcluster.V2Getter{}
		} else {
			c.TCGetter = tidbcluster.NewV1Getter(nil)
		}
		res, err = updateDelete(ctx, c, d.Destroy(ctx))
		if err == nil {
			r.CachedProvider.Deregister(ctx, &cluster)
		}
	}

	if r.Config.DryRun.Enabled {
		if err == nil {
			r.MaxTimes.Done(req.NamespacedName.String())
			clusterDryrunMetricsLabels["result"] = "passed"
			metrics.ClusterDryrun.With(clusterDryrunMetricsLabels).Set(1)
		} else {
			clusterDryrunMetricsLabels["result"] = "failed"
			metrics.ClusterDryrun.With(clusterDryrunMetricsLabels).Set(1)
		}
	}
	if err == nil {
		r.EventRecorder.Event(&cluster, corev1.EventTypeNormal, "ReconcileSuccess", "")
	} else {
		r.EventRecorder.Event(&cluster, corev1.EventTypeWarning, "ReconcileFail", err.Error())
	}

	return res, err
}

// TODO(chenfei): [feature annotation migration]remove this later
func (r *ClusterReconciler) migrateFeatureAnnotations(ctx context.Context, t *task.Context) {
	cluster := t.Cluster
	if cluster.Annotations == nil {
		cluster.Annotations = make(map[string]string)
	}

	if value, ok := cluster.Annotations[consts.KMSEnabledKey]; ok {
		featKey := cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseAtRestEncryption
		if _, ok := cluster.Annotations[featKey]; !ok {
			cluster.Annotations[featKey] = value
		}
	}

	if value, ok := cluster.Annotations[consts.CMEKEnabledKey]; ok {
		featKey := cloudv1.AnnoKeyFeatureNamePrefix + consts.ClusterFeatureUseCMEK
		if _, ok := cluster.Annotations[featKey]; !ok {
			cluster.Annotations[featKey] = value
		}
	}
	enableVolumeAttributesClassIfSupported := func() {
		if supportsVolumeAttributesClass(
			cluster.Spec.Kubernetes.Provider,
			t.GetNodeVersion(true),
			cluster.FeatureEnabled(consts.ClusterFeatureUseTiDBOperatorV2),
		) {
			enableIfNotExists(cluster, consts.ClusterFeatureUseVolumeAttributesClass)
		}
	}

	switch phase := t.Cluster.Status.Phase; phase {
	case cloudv1.ClusterPhasePending, "":
		// enable the following features by default. This is compatible logic for cloud-native clusters whose creation logic
		// doesn't pass through the cluster-server which is responsible for initializing features in create_cluster_utils.go:initClusterFeatures.
		enableIfNotExists(t.Cluster, consts.ClusterFeatureTiDBExposeReplayerPath)
		enableIfNotExists(t.Cluster, consts.ClusterFeatureTiDBExposeOOMPath)
		enableIfNotExists(t.Cluster, consts.ClusterFeatureDedicatedSG)

		// set auditlog sidecar image tag annotation for new clusters.
		if _, ok := t.Cluster.Annotations[consts.AuditLogSidecarImageTag]; !ok && r.Config.ClusterController.Audit.DefaultTag != "" {
			t.Cluster.Annotations[consts.AuditLogSidecarImageTag] = r.Config.ClusterController.Audit.DefaultTag
		}

		// Only enable the feature for GCP for now.
		if t.Cluster.Spec.Kubernetes.Provider == cloudv1.ProviderTypeGKE {
			enableIfNotExists(t.Cluster, consts.ClusterFeatureEnableGracefulShutdown)
			enableIfNotExists(t.Cluster, consts.ClusterFeatureRemoveNEGAnnotation)
		}

		if config.FeatureEnabled(r.Config.FeatureGates, config.SetPriorityClass, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) {
			enableIfNotExists(t.Cluster, consts.ClusterFeatureSetPriorityClass)
		}
		if config.FeatureEnabled(r.Config.FeatureGates, config.UseTiDBOperatorV2, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) {
			enableIfNotExists(t.Cluster, consts.ClusterFeatureUseTiDBOperatorV2)
			enableIfNotExists(t.Cluster, consts.ClusterFeatureDisableV2PDReadinessProbe)
		}

	case cloudv1.ClusterPhaseCreating:
		if config.FeatureEnabled(r.Config.FeatureGates, config.UseVolumeAttributesClass, cluster.GetTenantID(), cluster.GetProjectID()) {
			enableVolumeAttributesClassIfSupported()
		}

	case cloudv1.ClusterPhaseRunning:
		// Enable supported features for existing GKE clusters.
		if t.Cluster.Spec.Kubernetes.Provider == cloudv1.ProviderTypeGKE {
			if config.FeatureEnabled(r.Config.FeatureGates, config.RemoveNEGAnnotationForExisting, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) {
				enableIfNotExists(t.Cluster, consts.ClusterFeatureRemoveNEGAnnotation)
			}
			if config.FeatureEnabled(r.Config.FeatureGates, config.EnableGracefulShutdownForExisting, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) {
				enableIfNotExists(t.Cluster, consts.ClusterFeatureEnableGracefulShutdown)
			}
			enableVolumeAttributesClassIfSupported()
		}
	}
}

func enableIfNotExists(cluster *cloudv1.Cluster, featureName string) {
	featKey := cloudv1.AnnoKeyFeatureNamePrefix + featureName
	if _, ok := cluster.Annotations[featKey]; !ok {
		cluster.EnableFeature(featureName)
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	r.httpClient = &http.Client{
		Timeout: 10 * time.Second,
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{
				// nolint: gosec
				InsecureSkipVerify: true,
			},
			IdleConnTimeout: 30 * time.Second,
		},
	}

	// for now, only used for tokenBasedAuth secrets need to be watched
	secretPredicate, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			cloudv1.LabelUsedFor:     cloudv1.UsedForTokenBasedAuth,
			cloudv1.LabelServiceType: cloudv1.ServiceTypeDedicated,
		},
	})
	if err != nil {
		return fmt.Errorf("build secret predicate: %w", err)
	}

	// 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 err
	}

	if err = mgr.GetFieldIndexer().IndexField(context.TODO(), &cloudv1.Cluster{}, "spec.kubernetes.name", func(obj client.Object) []string {
		clusterObj := obj.(*cloudv1.Cluster)
		return []string{clusterObj.Spec.Kubernetes.Name}
	}); err != nil {
		return err
	}

	if r.Config.DryRun.Enabled {
		if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
			cl := cloudv1.ClusterList{}
			if err := mgr.GetClient().List(context.Background(), &cl); err != nil {
				r.Logger.Error(err, "unable to list clusters")
				return err
			}

			keys := []string{}
			for _, item := range cl.Items {
				keys = append(keys, client.ObjectKeyFromObject(&item).String())
			}
			r.MaxTimes.Register(3, keys...)
			r.MaxTimes.Done("dummy-key-not-exist") // manually do Done to trigger finish if keys is empty

			return nil
		})); err != nil {
			return err
		}
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Cluster{}, builder.WithPredicates(pds...)).
		Owns(&corev1.Secret{}, builder.WithPredicates(secretPredicate)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: controllers.GetMaxConcurrentReconciles(r.Config, "cluster")}).
		Complete(r)
}