package backup

import (
	"fmt"
	"path/filepath"
	"time"

	"reflect"

	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	"github.com/pingcap/tidb-operator/pkg/util"
	"github.com/pingcap/tidb-operator/pkg/util/errors"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	batchv1 "k8s.io/api/batch/v1"
	"k8s.io/api/batch/v1beta1"
	batchv1beta1 "k8s.io/api/batch/v1beta1"
	batchv2alpha1 "k8s.io/api/batch/v2alpha1"
	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/labels"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
	utilversion "k8s.io/kubernetes/pkg/util/version"
)

var (
	standardVersionStr = "v1.8"
	keyFunc            = cache.DeletionHandlingMetaNamespaceKeyFunc
	// ControllerKind contains the schema.GroupVersionKind for this controller type
	ControllerKind = v1.SchemeGroupVersion.WithKind("TidbClusterBackup")
)

const (
	// DefaultBackupImage is the default backup image，if the user does not specify a backup image, it is used by default.
	DefaultBackupImage = "pingcap/tidb-backup:v1.0"

	// DataPathName is the name of the volume.
	DataPathName = "datapath"

	// DataMountPath is the directory for the data.
	DataMountPath = "/var/tidb/data"

	// BackupStoreDir is the backup directory for storing backup data.
	BackupStoreDir = "backup"

	// BackupContainerName is the backup container name in backup pod.
	BackupContainerName = "backup"

	// FullBackupRetryCount is the max retry count of full backup job when failed.
	FullBackupRetryCount int32 = 3
)

// Controller is a controller for managing TidbClusterBackup.
type Controller struct {
	cli        versioned.Interface
	kubeCli    kubernetes.Interface
	controller cache.Controller
	store      cache.Store
	queue      *workqueue.Type
}

// NewController returns a *TiDBClusterBackupController
func NewController(cli versioned.Interface, kubeCli kubernetes.Interface) *Controller {
	tcbc := &Controller{
		cli:     cli,
		kubeCli: kubeCli,
		queue:   workqueue.New(),
	}
	tcbc.store, tcbc.controller = cache.NewInformer(
		&cache.ListWatch{
			ListFunc: cache.ListFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return tcbc.cli.PingcapV1().TidbClusterBackups(metav1.NamespaceAll).List(opts)
			}),
			WatchFunc: cache.WatchFunc(func(opts metav1.ListOptions) (watch.Interface, error) {
				return tcbc.cli.PingcapV1().TidbClusterBackups(metav1.NamespaceAll).Watch(opts)
			}),
		},
		&v1.TidbClusterBackup{},
		30*time.Second,
		cache.ResourceEventHandlerFuncs{
			AddFunc:    tcbc.enqueueTidbClusterBackup,
			UpdateFunc: tcbc.updateTidbClusterBackup,
			DeleteFunc: tcbc.enqueueTidbClusterBackup,
		})
	return tcbc
}

func (tcbc *Controller) enqueueTidbClusterBackup(obj interface{}) {
	key, err := keyFunc(obj)
	if err != nil {
		log.Errorf("can't get key for obj: %v, error: %v", obj, err)
		return
	}

	tcbc.queue.Add(key)
}

// TODO: determine whether it's necessary to update tidbclusterBackup
func (tcbc *Controller) updateTidbClusterBackup(old, cur interface{}) {
	tcbc.enqueueTidbClusterBackup(cur)
}

// Run start the controller.
func (tcbc *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tcbc.queue.ShutDown()
	log.Infof("Starting TidbClusterBackup controller")
	go tcbc.controller.Run(wait.NeverStop)

	for i := 0; i < workers; i++ {
		go wait.Until(tcbc.worker, time.Second, stopCh)
	}

	<-stopCh
	log.Infof("Shutting down TidbClusterBackup controller")
}

func (tcbc *Controller) worker() {
	for {
		func() {
			key, quit := tcbc.queue.Get()
			if quit {
				return
			}
			defer tcbc.queue.Done(key)
			if err := tcbc.syncTidbClusterBackup(key.(string)); err != nil {
				log.Error(err)
			}
		}()
	}
}

func (tcbc *Controller) syncTidbClusterBackup(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing TidbClusterBackup[%s] (%v)", key, time.Since(startTime))
	}()

	obj, exists, storeErr := tcbc.store.GetByKey(key)
	if storeErr != nil {
		return errors.Logf(storeErr, "try to get obj: %s return error", key)
	}

	ns, tcbName, nsErr := cache.SplitMetaNamespaceKey(key)
	if nsErr != nil {
		return errors.Logf(nsErr, "error splitting TidbClusterBackup key: %q", key)
	}

	if !exists {
		log.Infof("TidbClusterBackup [%s] doesn't exist in store", key)
		// make sure tidbClusterBackup is actually deleted
		backups := tcbc.cli.PingcapV1().TidbClusterBackups(ns)
		if _, err := backups.Get(tcbName, metav1.GetOptions{}); err != nil {
			if apierrors.IsNotFound(err) {
				log.Infof("TidbClusterBackup [%s] has been deleted", key)
				//TODO try to clean backup Tasks created by the TidbClusterBackup
				if delErr := tcbc.deleteBackupTasks(ns, tcbName); delErr != nil {
					return errors.Logf(err, "TidbClusterBackup: [%s/%s]'s backup tasks deleted failed", ns, tcbName)
				}
				return nil
			}
			return errors.Trace(err)
		}
	}

	tcb, typeAssertionOk := obj.(*v1.TidbClusterBackup)
	if !typeAssertionOk {
		return errors.Errorf("obj: %v is not a *TidbClusterBackup", obj)
	}

	if tcb.Spec.Cluster == "" {
		return errors.Errorf("TidbClusterBackup: [%s/%s]'s Spec.Cluster is nil. TidbClusterBackup must specify a TidbClusterName", ns, tcb.ClusterName)
	}

	clusters := tcbc.cli.PingcapV1().TidbClusters(ns)
	if _, err := clusters.Get(tcb.Spec.Cluster, metav1.GetOptions{}); err != nil {
		if apierrors.IsNotFound(err) {
			log.Errorf("TidbClusterBackup: [%s/%s] 's TidbCluster: [%s/%s] does not exist, return err:%v", ns, tcbName, ns, tcb.Spec.Cluster, err)
		}
		return errors.Trace(err)
	}

	tcName := tcb.Spec.Cluster
	if tcb.Spec.PhysicalBackup != nil {
		backupPVs, err := tcbc.findBackupPVs(tcb)
		if err != nil {
			return errors.Trace(err)
		}
		timestamp := time.Now().Format("20060102150405")
		//TODO make sure tikvs have stop before create backup tasks
		for _, pv := range backupPVs {

			/*annPvTombstoned := pv.Annotations[volumeprovisioner.AnnPVTombstoned]
			if annPvTombstoned != "" {
				log.Infof("TidbClusterBackup:[%s/%s]: TidbCluster: [%s] 's pv:[%s] is Tomstone!", ns, tcbName, tcName, pv.GetName())
				continue
			}*/

			storeID, exist := pv.Labels[label.StoreIDLabelKey]
			if !exist {
				log.Errorf("TidbClusterBackup:[%s/%s]: TidbCluster: [%s] 's pv:[%s] does not have storeId!", ns, tcbName, tcName, pv.GetName())
				continue
			}

			if backupTaskStatus, ok := tcb.Status.PhysicalBackupStatus.BackupTaskStatuses[storeID]; ok && len(backupTaskStatus.TaskName) > 0 {
				log.Infof("TidbClusterBackup:[%s/%s], TidbCluster:[%s],store:[%s]'s Backup Task has created,it's state is %s", ns, tcbName, tcb.Spec.Cluster, storeID, backupTaskStatus.State)
				continue
			}

			if pv.Spec.HostPath == nil {
				return fmt.Errorf("TidbClusterBackup: [%s/%s], pv: [%s]'s Spec.HostPath is empty", ns, tcbName, pv.GetName())
			}

			if _, exist := pv.Annotations[volumeprovisioner.AnnProvisionerNode]; !exist {
				return fmt.Errorf("TidbClusterBackup: [%s/%s], pv: [%s]'s Annotations[AnnProvisionerNode] is empty", ns, tcbName, pv.GetName())
			}

			backupDirName := fmt.Sprintf("%s_%s_%s_%s", ns, tcb.Spec.Cluster, storeID, timestamp)
			pod := tcbc.newPhysicalBackupPod(tcb, &pv, backupDirName)

			if _, err := tcbc.kubeCli.CoreV1().Pods(pod.Namespace).Create(pod); err != nil {
				log.Error(err)
				continue
			}

		}
		return tcbc.syncPhysicalBackupStatus(tcb)
	}

	if tcb.Spec.FullBackup != nil {
		if len(tcb.Spec.FullBackup.BackupNode) == 0 || len(tcb.Spec.FullBackup.BackupDir) == 0 {
			return errors.Errorf("TidbClusterBackup: [%s/%s]'s FullBackup.BackupNode and FullBackup.BackupDir can't be empty", ns, tcbName)
		}
		cronJobName := fullBackupCronJobName(tcbName)

		if tcbc.versionLessThan(standardVersionStr) {
			newCronJob, err := tcbc.newFullBackupCronJobV2alpha1(tcb)
			if err != nil {
				return errors.Logf(err, "TidbClusterBackup :[%s/%s] failed to new full backup's CronJob", ns, tcb.Name)
			}
			cronJobs := tcbc.kubeCli.BatchV2alpha1().CronJobs(ns)
			cronJob, err := cronJobs.Get(cronJobName, metav1.GetOptions{})
			if err != nil {
				if !apierrors.IsNotFound(err) {
					return errors.Logf(err, "Get TidbClusterBackup: [%s/%s] 's cronJob: [%s] failed", ns, tcbName, cronJobName)
				}
				if _, err := cronJobs.Create(newCronJob); err != nil {
					if !apierrors.IsAlreadyExists(err) {
						return errors.Logf(err, "TidbClusterBackup: [%s/%s] create full backup cronJob failed", ns, tcbName)
					}
				}
				log.Infof("TidbClusterBackup: [%s/%s] create full backup cronJob successfully", ns, tcbName)
			}

			if !reflect.DeepEqual(&newCronJob.Spec, &cronJob.Spec) {
				if _, err := cronJobs.Update(newCronJob); err != nil {
					return errors.Logf(err, "Update TidbClusterBackup: [%s/%s] 's cronJob: [%s] failed", ns, tcbName, cronJobName)
				}
				log.Infof("Update TidbClusterBackup: [%s/%s] 's cronJob: [%s] successfully", ns, tcbName, cronJobName)
			}
		} else {
			newCronJob, err := tcbc.newFullBackupCronJobV1beta1(tcb)
			if err != nil {
				return errors.Logf(err, "TidbClusterBackup :[%s/%s] failed to new full backup's CronJob", ns, tcb.Name)
			}
			cronJob, err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).Get(cronJobName, metav1.GetOptions{})
			if err != nil {
				if !apierrors.IsNotFound(err) {
					return errors.Logf(err, "Get TidbClusterBackup: [%s/%s] 's cronJob: [%s] failed", ns, tcbName, cronJobName)
				}
				_, err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).Create(newCronJob)
				if err != nil {
					if !apierrors.IsAlreadyExists(err) {
						return errors.Logf(err, "TidbClusterBackup: [%s/%s] create full backup cronJob failed", ns, tcbName)
					}
				}
				log.Infof("TidbClusterBackup: [%s/%s] create full backup cronJob successfully", ns, tcbName)
			}

			if !reflect.DeepEqual(&newCronJob.Spec, &cronJob.Spec) {
				if _, err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).Update(newCronJob); err != nil {
					return errors.Logf(err, "Update TidbClusterBackup: [%s/%s] 's cronJob: [%s] failed", ns, tcbName, cronJobName)
				}
				log.Infof("Update TidbClusterBackup: [%s/%s] 's cronJob: [%s] successfully", ns, tcbName, cronJobName)
			}
		}

		// sync fulbackup status

		if err := tcbc.syncFullBackupStatus(tcb); err != nil {
			return errors.Logf(err, "TidbClusterBackup: [%s/%s] sync full backup status failed", ns, tcbName)
		}
		log.Infof("TidbClusterBackup: [%s/%s] sync full backup status successfully", ns, tcbName)

		// delete failed jobs,avoid restart pod continuously

		if err := tcbc.deleteFailedJob(tcb); err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

func (tcbc *Controller) newFullBackupCronJobV2alpha1(tcb *v1.TidbClusterBackup) (*batchv2alpha1.CronJob, error) {
	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()

	podSpec, err := tcbc.newFullBackupPodSpec(tcb)
	if err != nil {
		return nil, err
	}

	jobName := fullBackupCronJobName(tcbName)
	var filedJobsHistoryLimit int32 = 3
	var successfulJobsHistoryLimit int32 = 3
	var parallelism int32 = 1
	var completions int32 = 1
	return &batchv2alpha1.CronJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:      jobName,
			Namespace: ns,
			Labels:    backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
		},
		Spec: batchv2alpha1.CronJobSpec{
			Schedule:                   tcb.Spec.FullBackup.BackupCron,
			ConcurrencyPolicy:          batchv2alpha1.ForbidConcurrent,
			FailedJobsHistoryLimit:     &filedJobsHistoryLimit,
			SuccessfulJobsHistoryLimit: &successfulJobsHistoryLimit,
			Suspend:                    tcb.Spec.FullBackup.Suspend,
			JobTemplate: batchv2alpha1.JobTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
				},
				Spec: batchv1.JobSpec{
					Parallelism: &parallelism,
					Completions: &completions,
					Template: corev1.PodTemplateSpec{
						Spec: *podSpec,
					},
				},
			},
		},
	}, nil
}

func (tcbc *Controller) newFullBackupCronJobV1beta1(tcb *v1.TidbClusterBackup) (*v1beta1.CronJob, error) {
	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()
	podSpec, err := tcbc.newFullBackupPodSpec(tcb)
	if err != nil {
		return nil, err
	}

	jobName := fullBackupCronJobName(tcbName)
	var filedJobsHistoryLimit int32 = 3
	var successfulJobsHistoryLimit int32 = 3
	var parallelism int32 = 1
	var completions int32 = 1
	return &v1beta1.CronJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:      jobName,
			Namespace: ns,
			Labels:    backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
		},
		Spec: batchv1beta1.CronJobSpec{
			Schedule:                   tcb.Spec.FullBackup.BackupCron,
			ConcurrencyPolicy:          batchv1beta1.ForbidConcurrent,
			FailedJobsHistoryLimit:     &filedJobsHistoryLimit,
			SuccessfulJobsHistoryLimit: &successfulJobsHistoryLimit,
			Suspend:                    tcb.Spec.FullBackup.Suspend,
			JobTemplate: batchv1beta1.JobTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
				},
				Spec: batchv1.JobSpec{
					Parallelism: &parallelism,
					Completions: &completions,
					Template: corev1.PodTemplateSpec{
						Spec: *podSpec,
					},
				},
			},
		},
	}, nil
}

func (tcbc *Controller) newFullBackupPodSpec(tcb *v1.TidbClusterBackup) (*corev1.PodSpec, error) {
	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()
	tcName := tcb.Spec.Cluster

	pdURL := fmt.Sprintf("http://%s-pd.%s:2379", tcName, ns)

	params, err := tcbc.buildMydumperParams(tcb)
	if err != nil {
		log.Errorf("build TidbClusterBackup: [%s/%s] mydumper params failed,%v", ns, tcbName, err)
		return nil, err
	}

	backupDir := filepath.Join(tcb.Spec.FullBackup.BackupDir, tcName)

	return &corev1.PodSpec{
		NodeName:      tcb.Spec.FullBackup.BackupNode,
		RestartPolicy: corev1.RestartPolicyOnFailure,
		Volumes: []corev1.Volume{
			{
				Name: BackupStoreDir,
				VolumeSource: corev1.VolumeSource{
					HostPath: &corev1.HostPathVolumeSource{
						Path: backupDir,
					},
				},
			},
		},
		Containers: []corev1.Container{
			{
				Name:    BackupContainerName,
				Image:   tcb.Spec.BackupImage,
				Command: []string{"/full_backup.sh"},
				Args:    params,
				VolumeMounts: []corev1.VolumeMount{
					{
						Name:      BackupStoreDir,
						MountPath: "/var/fullbackup",
					},
				},
				Env: []corev1.EnvVar{
					{
						Name: "BACKUP_NAME",
						ValueFrom: &corev1.EnvVarSource{
							FieldRef: &corev1.ObjectFieldSelector{
								FieldPath: "metadata.name",
							},
						},
					},
					{
						Name:  "PD_URL",
						Value: pdURL,
					},
				},
			},
		},
	}, nil
}

func (tcbc *Controller) syncFullBackupStatus(tcb *v1.TidbClusterBackup) error {
	newFullBackupStatus := v1.FullBackupStatus{}

	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()

	cronJobName := fullBackupCronJobName(tcbName)
	newFullBackupStatus.CronJobName = cronJobName
	if tcbc.versionLessThan(standardVersionStr) {
		cronJob, err := tcbc.kubeCli.BatchV2alpha1().CronJobs(ns).Get(cronJobName, metav1.GetOptions{})
		if err != nil {
			return errors.Trace(err)
		}
		if len(cronJob.Status.Active) != 0 {
			newFullBackupStatus.State = v1.BackupRunning
		} else {
			newFullBackupStatus.State = v1.BackupWaiting
		}
	} else {
		cronJob, err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).Get(cronJobName, metav1.GetOptions{})
		if err != nil {
			return errors.Trace(err)
		}
		if len(cronJob.Status.Active) != 0 {
			newFullBackupStatus.State = v1.BackupRunning
		} else {
			newFullBackupStatus.State = v1.BackupWaiting
		}
	}

	jobs, err := tcbc.kubeCli.BatchV1().Jobs(ns).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
		).String(),
	})

	if err != nil {
		return errors.Trace(err)
	}

	for _, job := range jobs.Items {
		pod, err := tcbc.getPodByJob(&job)
		if err != nil {
			return errors.Trace(err)
		}
		if pod == nil {
			continue
		}
		backupDir := filepath.Join(tcb.Spec.FullBackup.BackupDir, pod.GetName())

		backupTaskStatus := getBackupTaskStatus(pod)
		backupTaskStatus.BackupDir = backupDir

		if job.Status.Succeeded == *job.Spec.Completions {
			backupTaskStatus.State = v1.BackupSuccess
		}

		if newFullBackupStatus.BackupTaskStatuses == nil {
			newFullBackupStatus.BackupTaskStatuses = []v1.BackupTaskStatus{}
		}

		newFullBackupStatus.BackupTaskStatuses = append(newFullBackupStatus.BackupTaskStatuses, backupTaskStatus)
	}

	if !reflect.DeepEqual(&tcb.Status.FullBackupStatus, &newFullBackupStatus) {
		tcb.Status.FullBackupStatus = newFullBackupStatus
		if _, err := tcbc.cli.PingcapV1().TidbClusterBackups(ns).Update(tcb); err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

func (tcbc *Controller) getPodByJob(job *batchv1.Job) (*corev1.Pod, error) {
	pods, err := tcbc.kubeCli.CoreV1().Pods(job.GetNamespace()).List(
		metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				map[string]string{
					"job-name": job.GetName(),
				},
			).String(),
		})
	if err != nil {
		return nil, errors.Trace(err)
	}

	if len(pods.Items) > 0 {
		return &pods.Items[0], nil
	}

	return nil, nil
}

func (tcbc *Controller) deleteFailedJob(tcb *v1.TidbClusterBackup) error {
	tcbName := tcb.GetName()
	ns := tcb.GetNamespace()
	jobs, err := tcbc.kubeCli.BatchV1().Jobs(ns).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
		).String(),
	})
	if err != nil {
		return errors.Trace(err)
	}

	for _, job := range jobs.Items {
		if job.Status.Succeeded == *job.Spec.Completions {
			continue
		}

		pod, err := tcbc.getPodByJob(&job)
		if err != nil {
			return errors.Trace(err)
		}
		if pod == nil {
			continue
		}

		for _, cs := range pod.Status.ContainerStatuses {
			if cs.Name != BackupContainerName {
				continue
			}
			if cs.RestartCount > FullBackupRetryCount {
				err := tcbc.kubeCli.BatchV1().Jobs(ns).Delete(job.GetName(), &metav1.DeleteOptions{})
				if err != nil {
					return errors.Trace(err)
				}
			}
		}
	}

	return nil
}

func (tcbc *Controller) newPhysicalBackupPod(tcb *v1.TidbClusterBackup, pv *corev1.PersistentVolume, backupDirName string) *corev1.Pod {
	controller := true
	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()
	ownerRef := metav1.OwnerReference{
		APIVersion: ControllerKind.GroupVersion().String(),
		Kind:       ControllerKind.Kind,
		Name:       tcbName,
		UID:        tcb.UID,
		Controller: &controller,
	}
	backupImage := DefaultBackupImage
	if len(tcb.Spec.BackupImage) != 0 {
		backupImage = tcb.Spec.BackupImage
	}

	storeID := pv.Labels[label.StoreIDLabelKey]
	podName := physicalBackupTaskName(tcbName, storeID)

	srcPath := filepath.Join(DataMountPath, util.TiKVStoreDir)
	backupPath := filepath.Join(DataMountPath, BackupStoreDir, backupDirName)

	podLabels := backupTaskLabels(ns, tcbName, label.PhysicalBackupTypeLabelValue)
	podLabels[label.StoreIDLabelKey] = storeID
	podAnn := map[string]string{
		label.AnnBackupDirlKey: filepath.Join(pv.Spec.HostPath.Path, BackupStoreDir, backupDirName),
	}

	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:            podName,
			Namespace:       ns,
			OwnerReferences: []metav1.OwnerReference{ownerRef},
			Annotations:     podAnn,
			Labels:          podLabels,
		},

		Spec: corev1.PodSpec{
			Volumes: []corev1.Volume{
				{
					Name:         DataPathName,
					VolumeSource: corev1.VolumeSource{HostPath: &corev1.HostPathVolumeSource{Path: pv.Spec.HostPath.Path}},
				},
			},
			Containers: []corev1.Container{
				{
					Name:    BackupContainerName,
					Image:   backupImage,
					Command: []string{"/physical_backup.sh"},
					Args:    []string{srcPath, backupPath},
					VolumeMounts: []corev1.VolumeMount{
						{
							Name:      DataPathName,
							MountPath: DataMountPath,
						},
					},
				},
			},
			RestartPolicy: corev1.RestartPolicyNever,
			NodeName:      pv.Annotations[volumeprovisioner.AnnProvisionerNode],
		},
	}

	return pod
}

func (tcbc *Controller) syncPhysicalBackupStatus(tcb *v1.TidbClusterBackup) error {
	ns := tcb.GetNamespace()
	tcbName := tcb.GetName()
	newPhysicalBackupStatus := v1.PhysicalBackupStatus{
		BackupTaskStatuses: map[string]v1.BackupTaskStatus{},
	}

	pods, err := tcbc.kubeCli.CoreV1().Pods(ns).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			backupTaskLabels(ns, tcbName, label.PhysicalBackupTypeLabelValue),
		).String(),
	})
	if err != nil {
		return errors.Logf(err, "List TidbClusterBackup: [%s/%s]'s pods failed", ns, tcbName)
	}
	for _, pod := range pods.Items {
		storeID := pod.Labels[label.StoreIDLabelKey]
		backupDir := pod.Annotations[label.AnnBackupDirlKey]
		backupTaskStatus := getBackupTaskStatus(&pod)
		backupTaskStatus.BackupDir = backupDir
		newPhysicalBackupStatus.BackupTaskStatuses[storeID] = backupTaskStatus
	}

	if reflect.DeepEqual(&tcb.Status.PhysicalBackupStatus, &newPhysicalBackupStatus) {
		log.Infof("TidbClusterBackup: [%s/%s]'s status is not changed", ns, tcbName)
		return nil
	}

	tcb.Status.PhysicalBackupStatus = newPhysicalBackupStatus
	if _, err := tcbc.cli.PingcapV1().TidbClusterBackups(ns).Update(tcb); err != nil {
		return errors.Logf(err, "TidbClusterBackup: [%s/%s] update status failed", ns, tcbName)
	}
	log.Infof("TidbClusterBackup: [%s/%s] update status successfully", ns, tcbName)

	return nil
}

func (tcbc *Controller) deleteBackupTasks(ns, tcbName string) error {
	// delete physical backup task
	pods := tcbc.kubeCli.CoreV1().Pods(ns)
	backupPods, podsErr := pods.List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			backupTaskLabels(ns, tcbName, label.PhysicalBackupTypeLabelValue),
		).String(),
	})
	if podsErr != nil {
		return errors.Trace(podsErr)
	}

	for _, backupPod := range backupPods.Items {
		podName := backupPod.GetName()
		if err := pods.Delete(podName, &metav1.DeleteOptions{}); err != nil {
			return errors.Logf(err, "Failed to delete TidbClusterBackup: [%s/%s]'s physical backup pod: [%s/%s], error", ns, tcbName, ns, podName)
		}
	}

	//delete full backup task
	backupCronJobs, err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).List(metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			backupTaskLabels(ns, tcbName, label.FullBakcupTypeLabelValue),
		).String(),
	})
	if err != nil {
		return errors.Trace(err)
	}

	for _, cronJob := range backupCronJobs.Items {
		cronJobName := cronJob.GetName()
		err := tcbc.kubeCli.BatchV1beta1().CronJobs(ns).Delete(cronJobName, &metav1.DeleteOptions{})
		if err != nil {
			return errors.Logf(err, "Failed to delete TidbClusterBackup: [%s/%s]'s full backup cronJob: [%s/%s]", ns, tcbName, ns, cronJobName)
		}
		log.Infof(" Delete TidbClusterBackup: [%s/%s]'s CronJob: [%s/%s] successfully", ns, tcbName, ns, cronJob)
	}

	return nil
}

func (tcbc *Controller) findBackupPVs(tcb *v1.TidbClusterBackup) ([]corev1.PersistentVolume, error) {
	ns := tcb.GetNamespace()
	tcName := tcb.Spec.Cluster
	pvLabel := label.New().Namespace(ns).Cluster(tcName).App(label.TiKVLabelVal)
	pvListOptions := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(pvLabel.Labels()).String(),
	}
	pvList, err := tcbc.kubeCli.CoreV1().PersistentVolumes().List(pvListOptions)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if tcb.Spec.PhysicalBackup.BackupStores != nil && len(tcb.Spec.PhysicalBackup.BackupStores) > 0 {
		var backupPVs []corev1.PersistentVolume
		for _, storeID := range tcb.Spec.PhysicalBackup.BackupStores {
			backupPV := tcbc.findPVByStoreID(pvList.Items, storeID)
			if backupPV != nil {
				backupPVs = append(backupPVs, *backupPV)
			}
		}
		return backupPVs, nil
	}

	return pvList.Items, nil
}

func (tcbc *Controller) findPVByStoreID(pvs []corev1.PersistentVolume, storeID string) *corev1.PersistentVolume {
	for _, pv := range pvs {
		if id, ok := pv.Labels[label.StoreIDLabelKey]; ok && id == storeID {
			return &pv
		}
	}
	return nil
}

func physicalBackupTaskName(tcbName, storeID string) string {
	return fmt.Sprintf("%s-%s", tcbName, storeID)
}

func fullBackupCronJobName(tcbName string) string {
	return fmt.Sprintf("%s-fullbackup", tcbName)
}

func backupTaskLabels(ns, tidbClusterBackupName string, backupType string) map[string]string {
	labels := map[string]string{
		label.NamespaceLabelKey:  ns,
		label.BackupNameLabelKey: tidbClusterBackupName,
		label.BackupTypeLabelKey: backupType,
	}
	return labels
}

// parse pod's status
func getBackupTaskStatus(pod *corev1.Pod) v1.BackupTaskStatus {
	var message string

	var beginTime metav1.Time
	var endTime metav1.Time

	phase := pod.Status.Phase

	reason := pod.Status.Reason

	for _, container := range pod.Status.InitContainerStatuses {
		switch {
		case container.State.Terminated != nil && container.State.Terminated.ExitCode == 0:
			continue
		case container.State.Terminated != nil:
			// initialization is failed
			message = container.State.Terminated.Message
			if len(container.State.Terminated.Reason) == 0 {
				if container.State.Terminated.Signal != 0 {
					reason = fmt.Sprintf("Init:Signal:%d", container.State.Terminated.Signal)
				} else {
					reason = fmt.Sprintf("Init:ExitCode:%d", container.State.Terminated.ExitCode)
				}
			} else {
				reason = "Init:" + container.State.Terminated.Reason
			}
		case container.State.Waiting != nil && len(container.State.Waiting.Reason) > 0 && container.State.Waiting.Reason != "PodInitializing":
			reason = "Init:" + container.State.Waiting.Reason
			message = container.State.Waiting.Message
		default:
			reason = fmt.Sprintf("Init:%s %d", container.Name, len(pod.Spec.InitContainers))
		}
		break
	}

	for _, container := range pod.Status.ContainerStatuses {
		if container.Name != BackupContainerName {
			continue
		}
		if container.State.Terminated != nil {
			beginTime = container.State.Terminated.StartedAt
			endTime = container.State.Terminated.FinishedAt
			message = container.State.Terminated.Message
			if container.State.Terminated.Reason != "" {
				reason = container.State.Terminated.Reason
			} else {
				if container.State.Terminated.Signal != 0 {
					reason = fmt.Sprintf("Signal:%d", container.State.Terminated.Signal)
				} else {
					reason = fmt.Sprintf("ExitCode:%d", container.State.Terminated.ExitCode)
				}
			}
			continue
		}

		if container.State.Waiting != nil && container.State.Waiting.Reason != "" {
			reason = container.State.Waiting.Reason
			message = container.State.Waiting.Message
			continue
		}
	}

	if pod.DeletionTimestamp != nil && pod.Status.Reason == util.NodeUnreachablePodReason {
		reason = "Unknown"
	} else if pod.DeletionTimestamp != nil {
		reason = "Terminating"
	}
	backupStatus := v1.BackupTaskStatus{
		TaskName:  pod.GetName(),
		BeginTime: beginTime,
		EndTime:   endTime,
	}

	backupStatus.State = v1.BackupRunning
	if phase == corev1.PodSucceeded {
		backupStatus.State = v1.BackupSuccess
	}
	if phase == corev1.PodFailed {
		backupStatus.State = v1.BackupFailed
	}
	backupStatus.BackupNode = pod.Spec.NodeName
	backupStatus.Message = fmt.Sprintf("reason:%s,%s", reason, message)

	return backupStatus
}

func (tcbc *Controller) buildMydumperParams(tcb *v1.TidbClusterBackup) ([]string, error) {
	params := tcb.Spec.FullBackup.BackupParams
	if params == nil {
		params = map[string]string{}
	}

	tcName := tcb.Spec.Cluster
	ns := tcb.GetNamespace()
	svcName := fmt.Sprintf("%s-privileged-tidb", tcName)
	svc, err := tcbc.kubeCli.CoreV1().Services(ns).Get(svcName, metav1.GetOptions{})
	if err != nil {
		linkedErr := errors.Logf(err, "Get TidbCluster:[%s/%s]'s tidb service:[%s/%s] failed", ns, tcName, ns, svcName)
		return nil, linkedErr
	}
	params["--host"] = svc.Spec.ClusterIP
	params["--port"] = "4000"
	params["--user"] = "root"

	_, okLong := params["--threads"]
	_, okShort := params["-t"]
	if !okLong && !okShort {
		params["--threads"] = "3"
	}
	_, okLong = params["--rows"]
	_, okShort = params["-r"]
	if !okLong && !okShort {
		params["--rows"] = "10000"
	}

	paramsArr := []string{}

	for opt, val := range params {
		paramsArr = append(paramsArr, opt)
		paramsArr = append(paramsArr, val)
	}

	return paramsArr, nil
}

func (tcbc *Controller) versionLessThan(standardVersionStr string) bool {
	standardVersion, err := utilversion.ParseGeneric(standardVersionStr)
	if err != nil {
		log.Errorf("parse version: [%s] error:%v", standardVersionStr, err)
		return true
	}
	info, err := tcbc.kubeCli.Discovery().ServerVersion()
	if err != nil {
		log.Errorf("Discovery server version error :%v", err)
		return true
	}
	runtimeVersion, err := utilversion.ParseGeneric(info.GitVersion)
	if err != nil {
		log.Errorf("parse version: [%s] error:%v", info.GitVersion, err)
		return true
	}
	return runtimeVersion.LessThan(standardVersion)
}