package imports

import (
	"context"
	"fmt"
	"path"
	"strconv"
	"strings"
	"time"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"

	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/go-logr/logr"
	"github.com/pingcap/errors"
	configsdk "github.com/tidbcloud/config-service/sdk"

	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"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
	imports_pkg "github.com/tidbcloud/infra-api/pkg/imports"
	"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"
	generic_utils "github.com/tidbcloud/infra-api/pkg/util/generic"
)

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"

	AnnoKeyHibernateCommand        = "free-tier.tidbcloud.com/hibernate-command"
	AnnoToActiveHibernateCommand   = "ToActive"
	AnnoToInactiveHibernateCommand = "ToInactive"
)

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.Interface
}

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

//+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,
		},
	}

	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()}).
		WithEventFilter(utils.GetMigrationAnnotationSkipPredicator()).
		Complete(r)
}

type importConnector struct {
	client     client.Client
	envBuilder EnvBuilder
	configSvc  configsdk.Interface
	gitops     gitopsv2.Interface
}

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

func GetImportObjectGroup(ctx context.Context, infraClusterClient client.Client, i *cloudv1.Import) (*ImportAPIObjectGroup, error) {
	var c cloudv1.Cluster
	err := infraClusterClient.Get(ctx, types.NamespacedName{
		Name:      i.Spec.ClusterName,
		Namespace: i.Namespace,
	}, &c)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var n cloudv1.Network
	err = infraClusterClient.Get(ctx, types.NamespacedName{
		Name:      c.Spec.NetworkName,
		Namespace: c.Namespace,
	}, &n)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var p cloudv1.Provider
	err = infraClusterClient.Get(ctx, types.NamespacedName{
		Name:      n.Spec.Provider,
		Namespace: n.Namespace,
	}, &p)
	if err != nil {
		return nil, errors.Trace(err)
	}

	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)
	}

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

	env, err := ci.envBuilder.BuildEnvFromImport(ctx, i)
	if err != nil {
		return nil, errors.Trace(err)
	}
	importObjectGroup := env.GetImportObjectGroup()

	var c *cloudv1.Cluster = importObjectGroup.Cluster

	if !cloudv1.ClusterAvailable(*c) {
		if c.Status.Phase == cloudv1.ClusterPhaseHibernated {
			if c.Annotations == nil {
				c.Annotations = map[string]string{}
			}
			c.Annotations[AnnoKeyHibernateCommand] = AnnoToActiveHibernateCommand
		}
		if err := ci.client.Update(ctx, c); err != nil {
			return nil, err
		}
		return nil, errors.New(errClusterNotAvailable)
	}

	var k8sStatus *cloudv1.KubernetesStatus
	switch c.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeAWS:
		awsStatus, err := cloudv1.AwsClusterStatusFromCluster(c)
		if err != nil {
			return nil, err
		}
		k8sStatus = awsStatus.Kubernetes
	case consts.ProviderTypeGCP:
		gcpStatus, err := cloudv1.GcpClusterStatusFromCluster(c)
		if err != nil {
			return nil, err
		}
		k8sStatus = gcpStatus.Kubernetes
	case consts.ProviderTypeAWSFreeTier:
		awsFreeTierStatus, err := cloudv1.AwsFreeTierClusterStatusFromCluster(c)
		if err != nil {
			return nil, err
		}
		k8sStatus = awsFreeTierStatus.Kubernetes
	case consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		sharedTierStatus, err := cloudv1.SharedTierClusterStatusFromCluster(c)
		if err != nil {
			return nil, err
		}
		k8sStatus = sharedTierStatus.Kubernetes
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE:
		k8sStatus = &cloudv1.KubernetesStatus{Namespace: c.PrefixedId()}
	default:
		return nil, fmt.Errorf("unknown provider %s", c.Labels[cloudv1.ProviderTypeLabel])
	}
	if k8sStatus == nil {
		return nil, fmt.Errorf("k8s status is not yet set by provider %s", c.Labels[cloudv1.ProviderTypeLabel])
	}

	var n *cloudv1.Network = importObjectGroup.Network
	var p *cloudv1.Provider = importObjectGroup.Provider

	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{
		cli:        ci.client,
		clusterCli: clusterCli,
		restCfg:    restCfg,
		cluster:    c,
		network:    n,
		provider:   p,
		k8sStatus:  k8sStatus,
		env:        env,
		log:        log.FromContext(ctx),
		configSvc:  ci.configSvc,
		gitops:     ci.gitops,
	}, nil
}

type clusterImport struct {
	cli        client.Client
	clusterCli client.Client
	restCfg    *rest.Config
	cluster    *cloudv1.Cluster
	network    *cloudv1.Network
	provider   *cloudv1.Provider
	k8sStatus  *cloudv1.KubernetesStatus
	env        Env
	log        logr.Logger
	configSvc  configsdk.Interface
	gitops     gitopsv2.Interface
}

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.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.cli.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.cli.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.provider.Spec.Type, 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",
		},
	}
	if ci.provider.Spec.Type != consts.ProviderTypeAWSFreeTier {
		volumes = append(volumes, corev1.Volume{
			Name: "lightning-tls",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: tcName + "-cluster-client-secret",
				},
			},
		})
		mounts = append(mounts, corev1.VolumeMount{
			Name:      "lightning-tls",
			MountPath: "/var/lib/lightning-tls",
		})
	}

	var credEnv corev1.EnvVar

	lightningImage := defaultLightningImage

	switch ci.provider.Spec.Type {
	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.provider.Spec.Type == consts.ProviderTypeAWSFreeTier || ci.provider.Spec.Type == 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, consts.ProviderTypeGKE:
		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)
	}

	initSH := fmt.Sprintf(`
trap exit TERM # because we don't use dumb-init
host=${CLUSTER_NAME:=db}-tidb
port=%d

j=0
while [ $j -le 90 ]; do
  nc -zv -w 3 "$host" "$port"
  if [ $? -eq 0 ]; then
    echo "info: successfully connected to $host:$port, able to initialize now"
    exit 0
  else
    echo "info: failed to connect to $host:$port, sleep 1 second then retry"
      sleep 1
  fi
  j=$(( j + 1  ))
done

## continue to run lightning anyway, if still failed we can get the error from lightning
echo "failed to connect to $host:$port"
exit 0
	`, ci.cluster.GetPublicSQLEndpointPort())

	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: ns,
			Labels:    lightningLabels(name, tcName),
		},
		Spec: batchv1.JobSpec{
			BackoffLimit: pointer.Int32Ptr(0),
			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.
					InitContainers: []corev1.Container{
						{
							Name:            "tidb-lightning-init",
							Image:           "busybox:1.31.1",
							ImagePullPolicy: corev1.PullIfNotPresent,
							Command: []string{
								"sh",
								"-c",
								initSH,
							},
							Env: []corev1.EnvVar{{
								Name:  "CLUSTER_NAME",
								Value: tcName,
							}},
						},
					},
					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",
								"/dev/stdout",
							},
							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.ClusterName),
		attribute.String("import_name", i.Name),
	)

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

	if i.Spec.DryRun {
		err := ci.startPrecheck(ctx, i, cred)
		if err != nil {
			return err
		}
		return ci.syncPrecheckStatus(ctx, i)
	}

	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
	)
	if !isClusterShared(ci.provider.Spec.Type) {
		builder := &nodegroup.Builder{
			ConfigProvider: gitops.NewHardcode(),
		}
		region, err := ci.provider.FindRegion(ci.network.Spec.Region)
		if err != nil {
			return err
		}
		var machineType string
		switch ci.provider.Spec.Type {
		case consts.ProviderTypeAWS, consts.ProviderTypeEKS:
			machineType = defaultAWSMachineType
		case consts.ProviderTypeGCP, consts.ProviderTypeGKE:
			machineType = defaultGCPMachineType
		}

		// 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.Int64Ptr(1000),
			RunAsGroup: pointer.Int64Ptr(2000),
			FSGroup:    pointer.Int64Ptr(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",
		}
	}

	isDevTier := false
	if ci.provider.Spec.Type == consts.ProviderTypeAWSFreeTier {
		isDevTier = true
	}
	cfg, err := imports_pkg.GenerateLightningConfig(ci.gitops, ci.cluster, i, ci.env, isDevTier)
	if err != nil {
		return err
	}

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

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

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

	// 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 = generic_utils.MergeMap(pvc.Labels, labels)
		_, err := k8s.CreateIgnoreExists(ctx, ci.clusterCli, pvc)
		if err != nil {
			return err
		}
		appendVolume(job)
	} else {
		appendEmptyVolume(job)
	}

	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 = controllerutil.CreateOrUpdate(ctx, ci.clusterCli, svc, func() error {
		return nil
	}); err != nil {
		return err
	}

	i.Status.Phase = cloudv1.ImportPrepare
	return ci.cli.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)
	}
	if i.Spec.DryRun {
		return ci.syncPrecheckStatus(ctx, i)
	}

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

	// TODO: should be get task status after lightning pod running
	var link k8s.HttpLink
	var lightningClient LightningClient
	if !isClusterShared(ci.provider.Spec.Type) {
		cli, err := kubernetes.NewForConfig(ci.restCfg)
		if err != nil {
			return err
		}
		link = k8s.NewHttpLinkByKubeProxy(ctx, cli, ci.k8sStatus.Namespace, i.ObjectMeta.Name, "8289")
	} else {
		scheme := "http"
		if ci.provider.Spec.Type != consts.ProviderTypeAWSFreeTier {
			scheme = "https"
		}
		link = k8s.NewHTTPLinkNormal(ctx, scheme, ci.k8sStatus.Namespace, i.ObjectMeta.Name, "8289")
	}
	lightningClient = newLightningClient(link)
	ci.log.Info("Task info", "task", "start request to lightning")
	task, err := lightningClient.GetTaskProgress()
	if err != nil {
		// todo:error count

		ci.log.Error(err, "failed to get import task progress, requeue...", "import", i.ObjectMeta.Name)
		// This is expected to fail when the job just started.
		// return nil would also requeue but without exponential backoff
		return nil // nolint:nilerr
	}
	getCkptCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	tblCheckpointMap, err := lightningClient.GetAllTableCheckpoints(getCkptCtx, task, 4)
	if err != nil {
		ci.log.Error(err, "failed to get import table checkpoint map, ignore it and continue")
	}
	// todo:lightning is online,reset

	ci.log.Info("Task info", "taskMessage", task.Message)
	if task.Status == taskStatusNotStarted && !i.Status.IsTaskCreated {
		var cfg string
		cfg, err = ci.getLightningConfigFromSecret(ctx, i.ObjectMeta.Name)
		if err != nil {
			ci.log.Error(err, "Task create", "task", "get config failed before createTask", "import", i.ObjectMeta.Name)
			// todo: just alert,but report error to customer
			i.Status.Phase = cloudv1.ImportFailed
			i.Status.Message = "unknown error, please seek support"
			return ci.cli.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()
			ci.log.Error(err, "Task create", "task", "task create failed", "import", i.ObjectMeta.Name)
			return ci.cli.Status().Update(ctx, i)
		}
		i.Status.IsTaskCreated = true
		i.Status.TaskId = taskId
	}
	saveImportStatus(&(i.Status), task, tblCheckpointMap)
	if i.Status.Phase == cloudv1.ImportFailed {
		ci.log.Error(errors.New("task failed"), "Task info", "taskMessage", task.Message)
	}
	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
		}
	}
	ci.log.Info("Task info", "task", "update complete", "status", i.Status.Phase)
	return ci.cli.Status().Update(ctx, i)
}

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)
	}

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

	if i.Spec.DryRun {
		return ci.deletePrecheck(ctx, i)
	}

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

	if !isClusterShared(ci.provider.Spec.Type) {
		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.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 deleteK8SResource(ctx context.Context, cli client.Client, name types.NamespacedName, obj client.Object, opts ...client.DeleteOption) (reconciler.ExternalDeletion, error) {
	if err := cli.Get(ctx, name, obj); err != nil {
		if reconciler.IgnoreNotFound(err) != nil {
			return reconciler.ExternalDeletion{}, err
		}

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

	if !obj.GetDeletionTimestamp().IsZero() {
		return reconciler.ExternalDeletion{ResourceDeleted: false}, nil
	}

	return reconciler.ExternalDeletion{ResourceDeleted: false}, cli.Delete(ctx, obj, opts...)
}

func importErrorMessageDigest(message string) string {
	var substr string
	idx := strings.IndexAny(message, "\n")
	if idx > -1 {
		substr = message[:idx]
	} else {
		substr = message
	}
	return substr
}

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
}