package imports

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"path"
	"strconv"
	"strings"
	"time"

	"github.com/go-logr/logr"
	"github.com/pingcap/errors"
	plog "github.com/pingcap/log"
	configsdk "github.com/tidbcloud/config-service/sdk"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	batchv1 "k8s.io/api/batch/v1"
	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/kubernetes"
	"k8s.io/client-go/rest"
	"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/infra-provider/util/collections"
	"github.com/tidbcloud/infra-provider/util/config"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	"github.com/tidbcloud/infra-api/controllers/reconciler"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gitops"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-api/pkg/metrics/controller"
	"github.com/tidbcloud/infra-api/pkg/nodegroup"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
	"github.com/tidbcloud/infra-api/pkg/util"
	gitopsv2 "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
)

var tracer = otel.Tracer("ImportReconciler")

const (
	// TODO: it should be configured by gitops https://jira.tidbcloud.com/browse/DM-1688
	defaultLightningImage = "gcr.io/pingcap-public/dbaas/tidb-lightning:v5.3.0-20211123"
	gcpLightningImage     = "gcr.io/pingcap-public/dbaas/tidb-lightning:v5.3.0-20211123"

	DefaultStorageClassName = k8s.DefaultStorageClassName

	awsCredentialsKey  = "credentiascriptls"
	gcpCredentialsKey  = "privateKey"
	gcpCredentialsBase = "/root/gcp"
	gcpCredentialsPath = "key.json"
	lightningPort      = 8289
	lightningName      = "tidb-lightning"

	ImportFinalizer        = "import.cloud.pingcap.com/finalizer"
	errNotImport           = "not import resource"
	errClusterNotAvailable = "wait for cluster to be ready"

	defaultAWSMachineType = "c5.9xlarge"
	defaultGCPMachineType = "n2-standard-32"
)

var (
	cpuCoresMap = map[string]int64{
		defaultAWSMachineType: 36,
		defaultGCPMachineType: 32,
	}
	awsSpecialMachineTypeMap = map[string]string{
		"ap-southeast-5": "c6i.8xlarge",
	}
	awsClusterMachineTypeMap = map[string]string{
		"10601809957776321988": "c5.18xlarge", // for prod user
		"10128077518039658429": "c5.18xlarge", // for prod test
		"10254269139005784210": "c5.18xlarge", // for dev test
	}
)

type taskStatus uint8

const (
	taskStatusNotStarted taskStatus = 0
	taskStatusRunning    taskStatus = 1
	taskStatusCompleted  taskStatus = 2
)

// ImportReconciler reconciles a Import object
type ImportReconciler struct {
	client.Client
	Log                       logr.Logger
	Scheme                    *runtime.Scheme
	EnvBuilder                EnvBuilder
	ConfigService             configsdk.Interface
	Gitops                    gitopsv2.CCInterface
	clusterProxyAuthDecorator func(*http.Request) error
}

func NewImportReconciler(mgr ctrl.Manager, log logr.Logger, envBuilder EnvBuilder, configService configsdk.Interface, clusterProxyAuthDecorator func(*http.Request) error) *ImportReconciler {
	return &ImportReconciler{
		Client:                    mgr.GetClient(),
		Log:                       log,
		Scheme:                    mgr.GetScheme(),
		EnvBuilder:                envBuilder,
		ConfigService:             configService,
		clusterProxyAuthDecorator: clusterProxyAuthDecorator,
	}
}

func NewImportReconcilerCompat(mgr ctrl.Manager, infraCli client.Client, log logr.Logger, envBuilder EnvBuilder, configService configsdk.Interface, clusterProxyAuthDecorator func(*http.Request) error) *ImportReconciler {
	return &ImportReconciler{
		Client:                    infraCli,
		Log:                       log,
		Scheme:                    mgr.GetScheme(),
		EnvBuilder:                envBuilder,
		ConfigService:             configService,
		clusterProxyAuthDecorator: clusterProxyAuthDecorator,
	}
}

// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=imports,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=imports/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=imports/finalizers,verbs=update

func (i *ImportReconciler) SetUpGitopsV2() *ImportReconciler {
	i.Gitops = &gitopsv2.Gitops{
		Cli: i.Client,
	}
	return i
}

func (r *ImportReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, errResult error) {
	start := time.Now()
	importReconciler := &reconciler.Reconciler{
		NewResource: func() client.Object {
			return &cloudv1.Import{}
		},
		Client: r.Client,
		ExternalObj: importConnector{
			client:                    r.Client,
			envBuilder:                r.EnvBuilder,
			configSvc:                 r.ConfigService,
			gitops:                    r.Gitops,
			clusterProxyAuthDecorator: r.clusterProxyAuthDecorator,
		},
	}

	logger := r.Log.WithValues("import", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)

	spanCtx, span := tracer.Start(ctx, "Import Reconcile")
	defer func() {
		tracerutil.EndSpan(span, errResult)
		metricsctrl.ReconcileDuration.WithLabelValues("import").Observe(time.Since(start).Seconds())
	}()

	span.SetAttributes(attribute.String("import_name", req.Name))

	return importReconciler.Reconcile(spanCtx, req, ImportFinalizer)
}

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

type importConnector struct {
	client                    client.Client
	envBuilder                EnvBuilder
	configSvc                 configsdk.Interface
	gitops                    gitopsv2.CCInterface
	clusterProxyAuthDecorator func(req *http.Request) error
}

type ImportAPIObjectGroup struct {
	Import   *cloudv1.Import
	Cluster  *cloudv1.Cluster
	Network  *cloudv1.Network
	Provider *cloudv1.Provider
}

func getNamespacedNameFromNetworkName(networkName string, defaultNamespace string) *types.NamespacedName {
	// if the networkName contains "/", it means it's a cross namespaced network.
	strs := strings.Split(networkName, "/")
	if len(strs) == 2 {
		return &types.NamespacedName{Namespace: strs[0], Name: strs[1]}
	}
	return &types.NamespacedName{Name: networkName, Namespace: defaultNamespace}
}

// GetImportObjectGroup get all related objects for import CR object.
// It tries its best to collect all the related objects, but it's not guaranteed to get all of them.
// So some of the returned objects may be nil.
// Users should check the returned objects' nil status if needed.
// This function will return error if something unexpected happens, for example, some mandatory annotations / labels are missing,
// so that we need to fix the problem ASAP.
func GetImportObjectGroup(ctx context.Context, infraClusterClient client.Client, i *cloudv1.Import) (*ImportAPIObjectGroup, error) {
	logger := log.FromContext(ctx)
	c := new(cloudv1.Cluster)
	err := infraClusterClient.Get(ctx, types.NamespacedName{
		Name:      i.Spec.ClusterName,
		Namespace: i.Namespace,
	}, c)
	if err != nil {
		logger.Error(err, "failed to get cluster CR object")
		// Just setting the cluster CR object to nil, which is only mandatory when creating an import task.
		c = nil
	}

	var n *cloudv1.Network
	var networkName *types.NamespacedName
	networkNameStr, ok := i.Annotations[cloudv1.ImportAnnotationKeyNetworkName]
	if !ok {
		return nil, errors.New("network name is not set in import CR object's annotation")
	}
	logger.Info("get network name from import CR object's annotation", "network-name", networkNameStr)
	// Use import CR's namespace as default namespace, because usually it's the same as cluster CR's namespace.
	networkName = getNamespacedNameFromNetworkName(networkNameStr, i.Namespace)
	n = new(cloudv1.Network)
	err = infraClusterClient.Get(ctx, *networkName, n)
	if err != nil {
		logger.Error(err, "failed to get network CR object")
		// Just setting network CR object to nil, which is mandatory when constructing a gardener env.
		n = nil
	}

	var p *cloudv1.Provider
	providerNameStr, err := GetProviderType(i)
	if err != nil {
		return nil, errors.Annotate(err, "get provider type failed")
	}
	p = new(cloudv1.Provider)
	if err := infraClusterClient.Get(ctx, types.NamespacedName{
		Name:      providerNameStr,
		Namespace: i.Namespace,
	}, p); err != nil {
		logger.Error(err, "failed to get provider CR object")
		// Just setting the provider CR object to nil, which is only mandatory when creating an import task for dedicated tier.
		p = nil
	}

	return &ImportAPIObjectGroup{
		Import:   i,
		Cluster:  c,
		Network:  n,
		Provider: p,
	}, nil
}

func (ci importConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) {
	ctx, span := tracer.Start(ctx, "importConnector.Connect")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	i, ok := obj.(*cloudv1.Import)
	if !ok {
		return nil, errors.New(errNotImport)
	}
	logger := log.FromContext(ctx)

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
	)

	providerType, err := GetProviderType(i)
	if err != nil {
		return nil, errors.Annotate(err, "get provider type failed")
	}
	env, err := ci.envBuilder.BuildEnvFromImport(ctx, i)
	if err != nil {
		return nil, errors.Trace(err)
	}
	importObjectGroup, err := GetImportObjectGroup(ctx, ci.client, i)
	if err != nil {
		return nil, errors.Annotate(err, "get import object group failed")
	}
	if providerType == consts.ProviderTypeAWSFreeTier {
		// init base cluster for farther call
		importObjectGroup.Cluster = &cloudv1.Cluster{ObjectMeta: metav1.ObjectMeta{
			Name:      i.Spec.ClusterName,
			Namespace: i.Namespace,
		}}
	}

	// get the import namespace in data plane cluster
	dataPlaneNamespace, err := env.GetDataPlaneNamespace()
	if err != nil {
		return nil, errors.Annotate(err, "get data plane namespace failed")
	}
	k8sStatus := &cloudv1.KubernetesStatus{Namespace: dataPlaneNamespace}

	var (
		clusterCli client.Client
		restCfg    *rest.Config
	)
	clusterCli, err = env.GetControllerClient(ctx)
	if err != nil {
		return nil, err
	}
	restCfg, err = env.GetKubeConfig(ctx)
	if err != nil {
		return nil, err
	}

	return &clusterImport{
		infraCli:                  ci.client,
		clusterCli:                clusterCli,
		restCfg:                   restCfg,
		cluster:                   importObjectGroup.Cluster,
		network:                   importObjectGroup.Network,
		provider:                  importObjectGroup.Provider,
		k8sStatus:                 k8sStatus,
		providerType:              providerType,
		env:                       env,
		log:                       logger,
		configSvc:                 ci.configSvc,
		gitops:                    ci.gitops,
		clusterProxyAuthDecorator: ci.clusterProxyAuthDecorator,
	}, nil
}

func GetProviderType(imp *cloudv1.Import) (string, error) {
	if imp == nil {
		return "", errors.New("import CR is nil")
	}
	providerTypeFromLabel, ok := imp.Labels[cloudv1.ProviderTypeLabel]
	if !ok {
		return "", errors.New("cannot get provider type label from the import CR")
	}
	return providerTypeFromLabel, nil
}

type clusterImport struct {
	infraCli                  client.Client
	clusterCli                client.Client
	restCfg                   *rest.Config
	cluster                   *cloudv1.Cluster
	network                   *cloudv1.Network
	provider                  *cloudv1.Provider
	providerType              string
	k8sStatus                 *cloudv1.KubernetesStatus
	env                       Env
	log                       logr.Logger
	configSvc                 configsdk.Interface
	gitops                    gitopsv2.CCInterface
	clusterProxyAuthDecorator func(req *http.Request) error
}

func isClusterShared(provider string) bool {
	switch provider {
	case consts.ProviderTypeAWS,
		consts.ProviderTypeGCP,
		consts.ProviderTypeEKS,
		consts.ProviderTypeGKE:
		return false
	case consts.ProviderTypeAWSFreeTier,
		consts.ProviderTypeAWSSharedTier,
		consts.ProviderTypeGCPSharedTier:
		return true
	default:
		return true
	}
}

func (ci *clusterImport) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterImport.Observe")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	i, ok := obj.(*cloudv1.Import)
	if !ok {
		return reconciler.ExternalObservation{}, errors.New(errNotImport)
	}

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
	)

	if i.Status.Phase == cloudv1.ImportCompleted || i.Status.Phase == cloudv1.ImportFailed {
		return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: true}, ci.infraCli.Status().Update(ctx, i)
	}

	// creation is not completed
	if i.Status.Phase == "" {
		// creation of Import is NOT atomic, so we should transit to Delete() even though Create() is not completed (semi-state)
		if !obj.GetDeletionTimestamp().IsZero() {
			return reconciler.ExternalObservation{ResourceExists: true}, nil
		}
		return reconciler.ExternalObservation{ResourceExists: false}, ci.infraCli.Update(ctx, i)
	}

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

func newLightning(
	ci *clusterImport,
	name string,
	nodeSelector map[string]string,
	podSecurityContext *corev1.PodSecurityContext,
	cfg string,
	cc *cloudv1.ClusterConfig,
	cred []byte,
) (*batchv1.Job, *corev1.Secret) {
	ns := ci.k8sStatus.Namespace
	tcName := ci.env.TCName(ci.cluster.Name)
	secret := newLightningSecret(ci.providerType, name, ns, tcName, cfg, cred)
	volumes := []corev1.Volume{
		{
			Name: "config",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: name,
					Items: []corev1.KeyToPath{
						{Key: "tidb-lightning.toml", Path: "tidb-lightning.toml"},
					},
				},
			},
		},
	}
	mounts := []corev1.VolumeMount{
		{
			Name:      "config",
			MountPath: "/etc/tidb-lightning",
		},
	}
	secretName := ci.env.GetClusterCertSecretName()
	if secretName != "" {
		volumes = append(volumes, corev1.Volume{
			Name: "lightning-tls",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: secretName,
				},
			},
		})
		mounts = append(mounts, corev1.VolumeMount{
			Name:      "lightning-tls",
			MountPath: "/var/lib/lightning-tls",
		})
	}
	// There is a known bug in Lightning: https://github.com/pingcap/tidb/issues/53001, and so the setting 'tls=false' in
	// [tidb-configuration:https://github.com/tidbcloud/tidb-configuration/blob/prod/versions/v7.5.4/base/lightning.toml#L35] doesn't
	// take effects anymore for versions <= v8.2.x. Besides, the Lighting will use TLS configurations in [security] section as the TLS configurations in [tidb.security]
	// by default if it's unset. See https://docs.pingcap.com/tidb/stable/tidb-lightning-configuration get more information.
	//
	// So, here we mount tidb-server cert secret and use the 'ca.crt' to verify tidb server cert only. By the way,
	// mounting CA cert and then to verify the server-side cert that is the best practice and we should use it.
	tidbServerCertSecretName := ci.env.GetTiDBServerCertSecretName()
	if tidbServerCertSecretName != "" {
		volumes = append(volumes, corev1.Volume{
			Name: "tidb-server-cert",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: tidbServerCertSecretName,
				},
			},
		})
		mounts = append(mounts, corev1.VolumeMount{
			Name:      "tidb-server-cert",
			MountPath: "/var/lib/tidb-server-tls",
		})
	}

	var credEnv corev1.EnvVar

	lightningImage := defaultLightningImage

	switch ci.providerType {
	case consts.ProviderTypeAWS, consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeEKS:
		if len(cred) > 0 {
			v := corev1.Volume{
				Name: "aws-credentials",
				VolumeSource: corev1.VolumeSource{
					Secret: &corev1.SecretVolumeSource{
						SecretName: name,
						Items: []corev1.KeyToPath{
							{Key: awsCredentialsKey, Path: "credentials"},
						},
					},
				},
			}

			MountPath := "/root/.aws"
			if ci.providerType == consts.ProviderTypeAWSFreeTier || ci.providerType == consts.ProviderTypeAWSSharedTier {
				MountPath = "/.aws"
				v.VolumeSource.Secret.Items[0].Path = "config"
			}

			volumes = append(volumes, v)

			mounts = append(mounts, corev1.VolumeMount{
				Name:      "aws-credentials",
				MountPath: MountPath,
			})

			credEnv = corev1.EnvVar{
				// see https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html
				// to use config in ~/.aws/credentials we need set this
				Name:  "AWS_SDK_LOAD_CONFIG",
				Value: "true",
			}
		}
	case consts.ProviderTypeGCP, consts.ProviderTypeGCPSharedTier:
		v := corev1.Volume{
			Name: "gcp-credentials",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: name,
					Items: []corev1.KeyToPath{
						{Key: gcpCredentialsKey, Path: gcpCredentialsPath},
					},
				},
			},
		}
		volumes = append(volumes, v)
		mounts = append(mounts, corev1.VolumeMount{
			Name:      "gcp-credentials",
			MountPath: gcpCredentialsBase,
		})

		credEnv = corev1.EnvVar{
			Name:  "GOOGLE_APPLICATION_CREDENTIALS",
			Value: path.Join(gcpCredentialsBase, gcpCredentialsPath),
		}
		lightningImage = gcpLightningImage
	}

	if cc.Spec.Lightning != nil {
		lightningImage = fmt.Sprintf("%s:%s", *cc.Spec.Lightning.BaseImage, *cc.Spec.Lightning.Tag)
	}

	var backoffLimit int32 = 0
	if ci.env.EnableCheckpoint() {
		backoffLimit = 3
	}
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: ns,
			Labels:    lightningLabels(name, tcName),
		},
		Spec: batchv1.JobSpec{
			BackoffLimit: &backoffLimit,
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Annotations: map[string]string{
						"prometheus.io/scrape": "true",
						"prometheus.io/port":   strconv.Itoa(lightningPort),
						"prometheus.io/path":   "/metrics",
					},
					Labels: lightningLabels(name, tcName),
				},
				Spec: corev1.PodSpec{
					HostNetwork:   true,
					DNSPolicy:     corev1.DNSClusterFirstWithHostNet,
					RestartPolicy: corev1.RestartPolicyNever,
					Tolerations: []corev1.Toleration{
						lightningToleration,
					},
					// sometimes after start lightning will meet issue about dns(do not know the detail reason):
					// dial tcp: lookup db-tidb on 100.64.0.10:53: no such host
					//
					// this just check that we can resolve the host before start lightning.

					// 2023/04/23(okjiang)
					// As the time elapsed, we can't reproduce this issue now. Let's remove this piece of code for now.
					// If we happen to encounter this issue again in the near future, we will investigate its root cause.
					// TODO: If we don't encounter it within two months, please remove these comments.
					Containers: []corev1.Container{
						{
							Name:            "tidb-lightning",
							Image:           lightningImage,
							ImagePullPolicy: corev1.PullIfNotPresent,
							Command: []string{
								"/tidb-lightning",
								"--server-mode",
								"--config",
								"/etc/tidb-lightning/tidb-lightning.toml",
								"--log-file",
								"-",
							},
							VolumeMounts: mounts,
						},
					},
					Volumes:         volumes,
					NodeSelector:    nodeSelector,       // this will be empty for shared clusters
					SecurityContext: podSecurityContext, // nil for non-shared clusters now
				},
			},
		},
	}

	if credEnv.Name != "" {
		// this env may be empty (e.g for free-tier and shared-tier), if we set it to the container,
		// "spec.template.spec.containers[0].env[0].name: Required value" error will raise.
		job.Spec.Template.Spec.Containers[0].Env = append(job.Spec.Template.Spec.Containers[0].Env, credEnv)
	}

	return job, secret
}

func (ci *clusterImport) Create(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterImport.Create")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	i, ok := obj.(*cloudv1.Import)
	if !ok {
		return errors.New(errNotImport)
	}

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
	)

	if ci.cluster == nil {
		return errors.New("cannot get cluster CR object")
	}

	cred, err := ci.env.Credentials(ctx)
	if err != nil {
		return err
	}

	var (
		nodeSelector       map[string]string          // empty for shared clusters
		podSecurityContext *corev1.PodSecurityContext // nil for non-shared clusters now
		podAnnotations     map[string]string          // empty for non-shared clusters now
		machineType        string
	)
	if !isClusterShared(ci.providerType) {
		if ci.provider == nil {
			return errors.New("provider CR object is nil")
		}
		builder := &nodegroup.Builder{
			ConfigProvider: gitops.NewHardcode(),
		}
		clusterRegion, ok := i.Labels[cloudv1.RegionLabel]
		if !ok {
			return errors.Errorf("cannot find the cluster region label in import CR: %s", cloudv1.RegionLabel)
		}
		region, err := ci.provider.FindRegion(clusterRegion)
		if err != nil {
			return err
		}
		switch ci.providerType {
		case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
			// if region is ap-southeast-5, we should use c6i.xlarge
			if specialMachineType, ok := awsClusterMachineTypeMap[i.Spec.ClusterName]; ok {
				machineType = specialMachineType
			} else if specialMachineType, ok := awsSpecialMachineTypeMap[clusterRegion]; ok {
				machineType = specialMachineType
			} else {
				machineType = defaultAWSMachineType
			}
		case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
			machineType = defaultGCPMachineType
		}

		logger := ci.log.WithName(i.Name).WithValues("namespace", ci.k8sStatus.Namespace)

		logger.Info("create node group with machine type", "machineType", machineType, "region", clusterRegion)

		// concat servicetype tag
		tags := cloudv1.TagsFromLabels(i.Labels)
		tags[common.ServiceType] = common.Dedicated

		ngSet, err := builder.Build(&nodegroup.NodeTopologySpec{
			Provider: ci.provider,
			Components: []cloudv1.ClusterComponent{
				&lightning{
					replicas:     1,
					instanceType: machineType,
				},
			},
			Zones:     []string{region.Zones[0]},
			Tags:      tags,
			OwnerName: util.FnvHash(i.ClusterScopeName()),
		}, false)
		if err != nil {
			return err
		}
		nodeSelector, err = ci.env.CreateNodeGroup(ctx, ngSet)
		if err != nil {
			return err
		}
	} else {
		podSecurityContext = &corev1.PodSecurityContext{
			RunAsUser:  pointer.Int64(1000),
			RunAsGroup: pointer.Int64(2000),
			FSGroup:    pointer.Int64(2000),
		}
		// do not evit workload pods with cluster-autoscaler, https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler
		podAnnotations = map[string]string{
			"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
		}
	}

	var cfg *config.GenericConfig
	if len(i.Spec.LightningConfig) == 0 {
		return errors.New("lightning config is empty")
	}
	impConfigTOML := new(cloudv1.TomlConfig)
	if err := impConfigTOML.UnmarshalTOML([]byte(i.Spec.LightningConfig)); err != nil {
		return errors.Annotate(err, "parse config TOML data error")
	}
	cfg = config.New(impConfigTOML.MP)

	tcName := ci.env.TCName(ci.cluster.Name)
	labels := utils.CopyLabelsForWorkloads(i.Labels)

	cc, err := ci.gitops.DescribeClusterConfiguration(ci.cluster)
	if err != nil {
		return err
	}

	// use the CA cert in 'db-tidb-server-secret' to verify the tidb server-side cert when connecting to tidb at port 4000.
	// We don't present the client cert/key, because it's not required for enabling TLS.
	if ci.env.GetTiDBServerCertSecretName() != "" {
		cfg.Set("tidb.security.ca-path", "/var/lib/tidb-server-tls/ca.crt")
	}

	// adjust lightning cfg for different providers' args
	if cpuCore, ok := cpuCoresMap[machineType]; ok {
		AdjustLightningConfig(cfg, ProviderBaseConfig{DataFormat: i.Spec.Source.Format, CpuCore: cpuCore})
	}
	cfgBytes, err := cfg.MarshalTOML()
	if err != nil {
		return err
	}

	job, secret := newLightning(ci, i.ObjectMeta.Name, nodeSelector, podSecurityContext, string(cfgBytes), cc, cred)
	err = ci.env.PreCreateJobHook(job)
	if err != nil {
		return err
	}
	svc := newLightningService(i.ObjectMeta.Name, ci.k8sStatus.Namespace, tcName)
	job.Labels = collections.MergeMaps(labels, job.Labels)
	job.Spec.Template.Labels = collections.MergeMaps(labels, job.Spec.Template.Labels)
	job.Spec.Template.Annotations = collections.MergeMaps(podAnnotations, job.Spec.Template.Annotations)
	secret.Labels = collections.MergeMaps(labels, secret.Labels)
	svc.Labels = collections.MergeMaps(labels, svc.Labels)

	// ensure storage class before create pvc
	err = ci.ensureStorageClassExists(ctx)
	if err != nil {
		return err
	}
	// if specified StorageRequest, we create a pvc requesting StorageRequest
	// and mount it as sort dir for lighting.
	// if not, we mount a EmptyDir for lightning, note when run as non-root, it will
	// not has permission to create '/etc/endpoint' if we do not create this EmptyDir and mount it at '/etc/endpoint'.
	if i.Spec.StorageRequest != "" {
		storageClassName := ci.env.StorageClassName()
		pvc := newPersistentVolumeClaim(i.ObjectMeta.Name, ci.k8sStatus.Namespace, tcName, storageClassName, i.Spec.StorageRequest)
		pvc.Labels = collections.MergeMaps(labels, pvc.Labels)
		_, err := k8s.CreateIgnoreExists(ctx, ci.clusterCli, pvc)
		if err != nil {
			return err
		}
		appendVolume(job)
	} else {
		appendEmptyVolume(job)
	}

	if err := ci.env.PreCreateSecretHook(secret); err != nil {
		return err
	}
	if _, err = controllerutil.CreateOrUpdate(ctx, ci.clusterCli, secret, func() error {
		return nil
	}); err != nil {
		return err
	}

	if _, err = controllerutil.CreateOrUpdate(ctx, ci.clusterCli, job, func() error {
		return nil
	}); err != nil {
		return err
	}

	if err = ci.env.PreCreateSvcHook(svc); err != nil {
		return err
	}
	if _, err = controllerutil.CreateOrUpdate(ctx, ci.clusterCli, svc, func() error {
		return nil
	}); err != nil {
		return err
	}

	i.Status.Phase = cloudv1.ImportPrepare
	return ci.infraCli.Status().Update(ctx, i)
}

// Update send task and update task Progress
func (ci *clusterImport) Update(ctx context.Context, obj client.Object) (errResult error) {
	ctx, span := tracer.Start(ctx, "clusterImport.Update")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	i, ok := obj.(*cloudv1.Import)
	if !ok {
		return errors.New(errNotImport)
	}

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
	)

	logger := ci.log.WithName(i.Name).WithValues("namespace", ci.k8sStatus.Namespace)

	lightningPods := &corev1.PodList{}
	if err := ci.clusterCli.List(ctx, lightningPods,
		client.MatchingLabels{"job-name": i.Name},
		client.InNamespace(ci.k8sStatus.Namespace)); err != nil {
		ci.log.Error(err, "faild to get lightning pods, requeue...")
	}
	if len(lightningPods.Items) == 0 {
		logger.Info("lightning pod not up, requeue...")
		return nil
	}
	if lightningPods.Items[0].Status.Phase == "Pending" {
		logger.Info("lightning pod not up, requeue...")
		return nil
	}
	var link k8s.HttpLink
	var lightningClient LightningClient
	switch ci.providerType {
	case consts.ProviderTypeAWS, consts.ProviderTypeGCP, consts.ProviderTypeEKS:
		cliSet, err := kubernetes.NewForConfig(ci.restCfg)
		if err != nil {
			return err
		}

		link = k8s.NewHttpLinkByKubeProxy(ctx, cliSet, ci.k8sStatus.Namespace, i.ObjectMeta.Name, "8289")
		plog.Info("cliset", zap.Any("cliset", cliSet), zap.Any("link", link))
	case consts.ProviderTypeGKE:
		if ci.network == nil {
			return errors.New("network CR is nil when trying to get cluster proxy link endpoint")
		}
		type withProxyBaseUrl struct {
			ProxyBaseUrl string `json:"proxyBaseUrl,omitempty"`
		}
		objWithProxyBaseUrl := withProxyBaseUrl{}
		if err := json.Unmarshal(ci.network.Status.FromProvider.Raw, &objWithProxyBaseUrl); err != nil {
			return errors.Annotate(err, "failed to unmarshal provider specific network status")
		}
		clusterID := i.Spec.ClusterName
		importID := strings.TrimPrefix(i.Name, "import-")
		plog.Info("GKE constructing cluster proxy http link",
			zap.String("cluster id", clusterID),
			zap.String("import id", importID),
		)
		link = k8s.NewHTTPLinkByClusterProxy(objWithProxyBaseUrl.ProxyBaseUrl, clusterID, importID, ci.clusterProxyAuthDecorator)
	default:
		scheme := "http"
		if ci.env.GetClusterCertSecretName() != "" {
			scheme = "https"
		}
		link = k8s.NewHTTPLinkNormal(ctx, scheme, ci.k8sStatus.Namespace, i.ObjectMeta.Name, "8289")
	}
	lightningClient = newLightningClient(link)
	logger.Info("Task info", "task", "start request to lightning", "k8s namespace", ci.k8sStatus.Namespace, "import name", i.ObjectMeta.Name)
	task, err := lightningClient.GetTaskProgress()
	if err != nil {
		// todo:error count
		logger.Error(err, "failed to get import task progress, requeue...")
		// This is expected to fail when the job just started.
		// return nil would also requeue but without exponential backoff

		// but when job failed, we should update import's status by job status
		if setStatusErr := ci.setStatusFailedOrCompleteByJob(ctx, i); setStatusErr != nil {
			logger.Error(setStatusErr, "failed setStatusByJob requeue...")
			return nil // nolint:nilerr
		}
		logger.Info("Task info", "task", "update complete", "status", i.Status.Phase)
		return ci.infraCli.Status().Update(ctx, i)
	}

	// todo:lightning is online,reset
	ci.log.Info("Task info", "taskMessage", task.Message)
	if task.Status == taskStatusNotStarted && (!i.Status.IsTaskCreated || ci.env.EnableCheckpoint()) {
		var cfg string
		cfg, err = ci.getLightningConfigFromSecret(ctx, i.ObjectMeta.Name)
		if err != nil {
			logger.Error(err, "Task create", "task", "get config failed before createTask")
			// todo: just alert,but report error to customer
			i.Status.Phase = cloudv1.ImportFailed
			i.Status.Message = "unknown error, please seek support"
			return ci.infraCli.Status().Update(ctx, i)
		}
		taskId, err := lightningClient.CreateTask(cfg)
		if err != nil {
			// if create task failed,termination
			i.Status.Phase = cloudv1.ImportFailed
			i.Status.Message = err.Error()
			logger.Error(err, "Task create", "task", "task create failed")
			return ci.infraCli.Status().Update(ctx, i)
		}
		i.Status.IsTaskCreated = true
		i.Status.TaskId = taskId
	}

	lightningBackend := func() string {
		defaultBackend := "local"
		cfgStr, err := ci.getLightningConfigFromSecret(ctx, i.ObjectMeta.Name)
		if err != nil {
			logger.Error(err, "Get lightning config from secret failed.")
			return defaultBackend
		}
		cfg := config.New(make(map[string]interface{}))
		if err := cfg.UnmarshalTOML([]byte(cfgStr)); err != nil {
			logger.Error(err, "Unmarshal config string into TOML failed")
			return defaultBackend
		}
		v := cfg.Get("tikv-importer.backend")
		if v == nil {
			return defaultBackend
		}
		lightningBackendStr, err := v.AsString()
		if err != nil {
			return defaultBackend
		}
		return lightningBackendStr
	}()
	saveImportStatus(&(i.Status), task, lightningBackend, logger)
	if i.Status.Phase == cloudv1.ImportFailed {
		logger.Error(errors.New("task failed"), "Task info", "taskMessage", task.Message)
	}

	if err := ci.setStatusFailedOrCompleteByJob(ctx, i); err != nil {
		return err
	}
	logger.Info("Task info", "task", "update complete", "status", i.Status.Phase)
	return ci.infraCli.Status().Update(ctx, i)
}

func (ci *clusterImport) setStatusFailedOrCompleteByJob(ctx context.Context, i *cloudv1.Import) error {
	var job batchv1.Job
	if err := ci.clusterCli.Get(ctx, types.NamespacedName{
		Name:      i.ObjectMeta.Name,
		Namespace: ci.k8sStatus.Namespace,
	}, &job); err != nil {
		return err
	}

	for _, cond := range job.Status.Conditions {
		if cond.Status != corev1.ConditionTrue {
			continue
		}

		if cond.Type == batchv1.JobFailed {
			i.Status.Phase = cloudv1.ImportFailed
			if i.Status.Message == "" {
				i.Status.Message = "abnormal termination, please seek support"
			}
		} else if cond.Type == batchv1.JobComplete {
			i.Status.Phase = cloudv1.ImportCompleted
		}
	}
	return nil
}

func (ci *clusterImport) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterImport.Delete")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	i, ok := obj.(*cloudv1.Import)
	if !ok {
		return reconciler.ExternalDeletion{}, errors.New(errNotImport)
	}

	// Retain the site for easy troubleshooting by Oncall
	if awsClusterMachineTypeMap[i.Spec.ClusterName] != "" {
		return reconciler.ExternalDeletion{ResourceDeleted: true}, nil
	}

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
		attribute.String("import_phase", string(i.Status.Phase)),
	)

	jobDeleted, err := ci.deleteJob(ctx, i)
	if err != nil {
		return reconciler.ExternalDeletion{}, err
	}

	if !isClusterShared(ci.providerType) {
		ngSet := &nodegroup.Set{
			Name: util.FnvHash(i.ClusterScopeName()),
		}
		if err := ci.env.DeleteNodeGroup(ctx, ngSet); err != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}
	return reconciler.ExternalDeletion{ResourceDeleted: jobDeleted.ResourceDeleted}, nil
}

func (ci *clusterImport) deleteJob(ctx context.Context, i *cloudv1.Import) (_ reconciler.ExternalDeletion, errResult error) {
	ctx, span := tracer.Start(ctx, "clusterImport.deleteJob")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	span.SetAttributes(
		attribute.String("cluster_name", i.Spec.ClusterName),
		attribute.String("import_name", i.Name),
		attribute.String("import_phase", string(i.Status.Phase)),
	)

	span.AddEvent("Delete import service")
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.ObjectMeta.Name,
			Namespace: ci.k8sStatus.Namespace,
		},
	}
	err := ci.clusterCli.Delete(ctx, svc)
	if err != nil {
		ci.log.Error(err, "Deleting import service failed", "service name", i.Name)
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}

	span.AddEvent("Delete import batch job")
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.ObjectMeta.Name,
			Namespace: ci.k8sStatus.Namespace,
		},
	}
	background := metav1.DeletePropagationBackground
	err = ci.clusterCli.Delete(ctx, job, &client.DeleteOptions{PropagationPolicy: &background})
	if err != nil {
		ci.log.Error(err, "Deleting import batch job failed", "batch job name", i.Name)
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}

	span.AddEvent("Delete import secret")
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.ObjectMeta.Name,
			Namespace: ci.k8sStatus.Namespace,
		},
	}
	err = ci.clusterCli.Delete(ctx, secret)
	if err != nil {
		ci.log.Error(err, "Deleting import secret failed", "secret name", i.Name)
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}

	span.AddEvent("Delete import pvc")
	pvc := &corev1.PersistentVolumeClaim{
		ObjectMeta: metav1.ObjectMeta{
			Name:      i.ObjectMeta.Name,
			Namespace: ci.k8sStatus.Namespace,
		},
	}
	err = ci.clusterCli.Delete(ctx, pvc)
	if err != nil {
		ci.log.Error(err, "Deleting import pvc failed", "pvc name", i.Name)
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}
	}

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

func (ci *clusterImport) getLightningConfigFromSecret(ctx context.Context, name string) (string, error) {
	secret := &corev1.Secret{}
	namespacedName := types.NamespacedName{
		Namespace: ci.k8sStatus.Namespace,
		Name:      name,
	}
	err := ci.clusterCli.Get(ctx, namespacedName, secret)
	if err != nil {
		return "", err
	}
	bytes := secret.Data[lightningConfigFileName]
	cfg := string(bytes)
	return cfg, nil
}

func (ci *clusterImport) ensureStorageClassExists(ctx context.Context) error {
	storageclass := ci.env.BuildStorageClass()
	// for garden cluster, we don't need to create storageclass
	if storageclass == nil {
		return nil
	}
	// if network task created,we ingnore the error.
	_, err := k8s.CreateIgnoreExists(ctx, ci.clusterCli, storageclass)
	return err
}