// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.package spec

package manager

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

	"github.com/juju/errors"
	"github.com/ngaut/log"
	pd "github.com/pingcap/pd/pd-client"
	tiapi "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/api"
	tictl "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/controller"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"

	"github.com/pingcap/private-tidb-operator/pkg/volumeprovisioner"

	batchv1 "k8s.io/api/batch/v1"
	"k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
)

func (m Manager) ListCluster(namespace string) ([]*Cluster, error) {
	instanceList, err := m.Cli.PingcapV1().TidbInstances(namespace).List(metav1.ListOptions{})
	if err != nil {
		log.Warnf("failed to get all clusters's info. error: %v", err)
		return nil, err
	}
	clusters := []*Cluster{}
	for _, item := range instanceList.Items {
		c := &Cluster{
			Name:      item.Namespace,
			CreatedAt: item.CreationTimestamp.Time,
			Pd:        &tiapi.PodSpec{Size: item.Spec.PD.Size},
			Tidb:      &tiapi.PodSpec{Size: item.Spec.TiDB.Size},
			Tikv:      &tiapi.PodSpec{Size: item.Spec.TiKV.Size},
			Deleted:   item.Spec.Deleted,
		}

		c.Config = item.Spec.Config
		if tictl.HasBinlogEnabled(&item) {
			c.KafkaAddr = item.Spec.KafkaAddr
		}
		clusters = append(clusters, c)
	}
	return clusters, nil
}

func (m Manager) GetCluster(namespace string) (*Cluster, error) {
	instance, err := m.Cli.PingcapV1().TidbInstances(namespace).Get(namespace, metav1.GetOptions{})
	if err != nil {
		if apierrors.IsNotFound(err) {
			log.Debugf("cluster %s/%s not found", namespace, clusterName)
			return nil, nil
		}
		log.Warnf("failed to get cluster %s/%s: %v", namespace, clusterName, err)
		return nil, err
	}
	if instance.Spec.Deleted {
		log.Infof("cluster %s/%s is graceful deleted", namespace, clusterName)
		return nil, nil
	}

	pdSetName := fmt.Sprintf("%s-pdset", clusterName)
	pdSet, err := m.Cli.PingcapV1().TidbSets(instance.Namespace).Get(pdSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("get cluster %s/%s 's pdset failed: %v", namespace, clusterName, err)
		return nil, err
	}

	tikvSetName := fmt.Sprintf("%s-tikvset", clusterName)
	tikvSet, err := m.Cli.PingcapV1().TidbSets(instance.Namespace).Get(tikvSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("get cluster %s/%s 's tikvset failed: %v", namespace, clusterName, err)
		return nil, err
	}

	tidbSetName := fmt.Sprintf("%s-tidbset", clusterName)
	tidbSet, err := m.Cli.PingcapV1().TidbSets(instance.Namespace).Get(tidbSetName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("get cluster %s/%s 's tidbset failed: %v", namespace, clusterName, err)
		return nil, err
	}

	c := &Cluster{
		Name:        instance.Namespace,
		CreatedAt:   instance.CreationTimestamp.Time,
		Pd:          makeRespPodSpec(tictl.PDPodType, instance),
		Tidb:        makeRespPodSpec(tictl.TiDBPodType, instance),
		Tikv:        makeRespPodSpec(tictl.TiKVPodType, instance),
		Initialized: instance.Status.Initialized,
	}

	//Set kafka address when binlog is enabled
	if tictl.HasBinlogEnabled(instance) {
		c.KafkaAddr = instance.Spec.KafkaAddr
	}

	l := label.New().Cluster(clusterName).Labels()
	option := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(l).String()}
	pods, err := m.Cli.CoreV1().Pods(namespace).List(option)
	if err != nil {
		return nil, err
	}
	for _, pod := range pods.Items {
		if pod.ObjectMeta.Labels != nil {
			switch pod.ObjectMeta.Labels[label.AppLabelKey] {
			case label.PDLabelVal:
				status := makePodStatus(&pod, tictl.PDContainerType, pdSet)
				c.PdStatus = append(c.PdStatus, status)
			case label.TiKVLabelVal:
				status := makePodStatus(&pod, tictl.TiKVContainerType, tikvSet)
				c.TikvStatus = append(c.TikvStatus, status)
			case label.TiDBLabelVal:
				status := makePodStatus(&pod, tictl.TiDBContainerType, tidbSet)
				c.TidbStatus = append(c.TidbStatus, status)
			case "tidb-monitor", "configure-grafana", "privileged-tidb":
				log.Debugf("tidb monitor: %s, namespace: %s", pod.ObjectMeta.Labels[label.AppLabelKey], namespace)
			default:
				log.Warnf("unexpected pod type: %s, namespace: %s", pod.ObjectMeta.Labels[label.AppLabelKey], namespace)
			}
		}
	}
	serviceNames := map[string]string{
		"grafana":    getName("grafana"),
		"prometheus": getName("prometheus"),
		"tidb":       getName("tidb"),
	}
	if tictl.ExposePrivilegedTiDBSVC {
		serviceNames["privileged-tidb"] = getName("privileged-tidb")
	}
	for kind, name := range serviceNames {
		svc, err := m.Cli.CoreV1().Services(namespace).Get(name, metav1.GetOptions{})
		if err != nil {
			log.Errorf("failed to get service: %s, namespace: %s, error: %v", name, namespace, err)
			continue
		}
		service := m.getService(svc)
		switch kind {
		case "grafana":
			c.GrafanaService = service
		case "prometheus":
			c.PrometheusService = service
		case "tidb":
			c.TidbService = service
		case "privileged-tidb":
			c.PrivilegedTiDBSerivice = service
		default:
			log.Warnf("unexpected service kind: %s, namespace: %s", kind, namespace)
		}
	}
	return c, nil
}

func (m Manager) CreateNamespace(namespace string) error {
	ns := &v1.Namespace{
		ObjectMeta: metav1.ObjectMeta{
			Name:   namespace,
			Labels: label.New().Cluster(clusterName).Labels(),
		},
	}
	_, err := m.Cli.CoreV1().Namespaces().Create(ns)
	if err != nil {
		log.Errorf("failed to create namespace: %s, error: %v", namespace, err)
	}
	return err
}

func (m Manager) CreateInstance(c *Cluster) error {
	ns := c.Name

	if len(c.Config) == 0 {
		return errors.Errorf("tidb cluster: [%s]'s Config is empty", ns)
	}

	_, err := m.Cli.CoreV1().ConfigMaps(m.Namespace).Get(c.Config, metav1.GetOptions{})
	if err != nil {
		errMsg := fmt.Sprintf("failed to get tidb cluster: [%s]'s configmap: [%s/%s], error: %v", ns, m.Namespace, c.Config, err)
		log.Error(errMsg)
		return errors.New(errMsg)
	}

	//create namespace with cluster name
	err = m.CreateNamespace(ns)
	if err != nil {
		return err
	}
	st := m.ServiceType
	switch c.ServiceType {
	case string(v1.ServiceTypeClusterIP), string(v1.ServiceTypeNodePort), string(v1.ServiceTypeLoadBalancer):
		st = c.ServiceType
	default:
		log.Warnf("wrong ServiceType %s for %s, use default %s", c.ServiceType, ns, m.ServiceType)
	}

	instance := &tiapi.TidbInstance{
		ObjectMeta: metav1.ObjectMeta{
			Name: ns,
		},
		Spec: tiapi.InstanceSpec{
			PD:                 m.revisePodSpec(tictl.PDPodType, c.Pd),
			TiKV:               m.revisePodSpec(tictl.TiKVPodType, c.Tikv),
			TiDB:               m.revisePodSpec(tictl.TiDBPodType, c.Tidb),
			Monitor:            m.revisePodSpec(tictl.MonitorPodType, c.Monitor),
			ServiceType:        st,
			TiDBLease:          c.TidbLease,
			MonitorReserveDays: c.MonitorReserveDays,
			RootPassword:       c.RootPassword,
			Labels:             c.Labels,
			Config:             c.Config,
			KafkaAddr:          c.KafkaAddr,
			Secrets:            c.Secrets,
		},
	}
	_, err = m.Cli.PingcapV1().TidbInstances(ns).Create(instance)
	if err != nil {
		log.Errorf("failed to create TidbInstance %s/%s: %v", ns, instance.Name, err)
		return err
	}
	go m.SetPassword(ns, c.RootPassword, true)
	return nil
}

func (m Manager) ResizeInstance(namespace string, c Cluster) error {
	instance, err := m.Cli.PingcapV1().TidbInstances(namespace).Get(namespace, metav1.GetOptions{})
	if err != nil {
		log.Errorf("failed to get TidbInstance %s/%s: %v", namespace, namespace, err)
		return errors.Trace(err)
	}

	log.Debugf("resize TidbCluster[%s/%s], pd: %d->%d, tikv: %d->%d, tidb: %d->%d",
		namespace, clusterName,
		instance.Spec.PD.Size, c.Pd.Size,
		instance.Spec.TiKV.Size, c.Tikv.Size,
		instance.Spec.TiDB.Size, c.Tidb.Size)

	if c.Pd.Size != 0 && c.Pd.Size != instance.Spec.PD.Size {
		instance.Spec.PD.Size = c.Pd.Size
	}
	if c.Tikv.Size != 0 && c.Tikv.Size != instance.Spec.TiKV.Size {
		instance.Spec.TiKV.Size = c.Tikv.Size
	}
	if c.Tidb.Size != 0 && c.Tidb.Size != instance.Spec.TiDB.Size {
		instance.Spec.TiDB.Size = c.Tidb.Size
	}

	_, err = m.Cli.PingcapV1().TidbInstances(namespace).Update(instance)
	if err != nil {
		log.Errorf("failed to resize TidbInstance %s/%s: %v", namespace, instance.GetName(), err)
		return errors.Trace(err)
	}
	log.Infof("TidbInstance[%s/%s] resized successfully", namespace, instance.GetName())
	return nil
}

func (m Manager) UpgradeInstance(namespace string, c Cluster) (bool, error) {
	var imageUpdate bool

	s, err := m.Cli.PingcapV1().TidbInstances(namespace).Get(namespace, metav1.GetOptions{})
	if err != nil {
		log.Errorf("failed to get TidbInstance %s/%s, update %t, %v", namespace, namespace, imageUpdate, err)
		return imageUpdate, errors.Trace(err)
	}

	if checkContainerSpec(tictl.PDContainerType, c.Pd) {
		m.revisePodSpec(tictl.PDPodType, c.Pd)
		newPdImage := tictl.GetImageByTiPodSpec(tictl.PDContainerType, c.Pd)
		oldPdImage := tictl.GetImageByTiPodSpec(tictl.PDContainerType, s.Spec.PD)
		if oldPdImage != newPdImage {
			imageUpdate = true
			tictl.SetImageByTiPodSpec(tictl.PDContainerType, s.Spec.PD, newPdImage)
			s.Status.PDStatus.UpgradeState = tiapi.Upgrading
			s.Status.PDStatus.UpgradeBeginTime = metav1.Now()
		}
	}

	if checkContainerSpec(tictl.TiKVContainerType, c.Tikv) {
		m.revisePodSpec(tictl.TiKVPodType, c.Tikv)
		newTikvImage := tictl.GetImageByTiPodSpec(tictl.TiKVContainerType, c.Tikv)
		oldTikvImage := tictl.GetImageByTiPodSpec(tictl.TiKVContainerType, s.Spec.TiKV)
		if oldTikvImage != newTikvImage {
			imageUpdate = true
			tictl.SetImageByTiPodSpec(tictl.TiKVContainerType, s.Spec.TiKV, newTikvImage)
			s.Status.TiKVStatus.UpgradeState = tiapi.Upgrading
			s.Status.TiKVStatus.UpgradeBeginTime = metav1.Now()
		}
	}

	if checkContainerSpec(tictl.BinlogContainerType, c.Tidb) {
		m.revisePodSpec(tictl.TiDBPodType, c.Tidb)
		if len(c.Tidb.Containers) > len(s.Spec.TiDB.Containers) {
			// add a new container
			imageUpdate = true
			binlogCSpec := getContainerSpec(tictl.BinlogContainerType, c.Tidb)
			s.Spec.TiDB.Containers = append(s.Spec.TiDB.Containers, binlogCSpec)
			s.Spec.KafkaAddr = c.KafkaAddr
			s.Status.TiDBStatus.UpgradeState = tiapi.Upgrading
			s.Status.TiDBStatus.UpgradeBeginTime = metav1.Now()
		} else {
			// binlog's image may has changed
			newBinlogImage := tictl.GetImageByTiPodSpec(tictl.BinlogContainerType, c.Tidb)
			oldBinlogImage := tictl.GetImageByTiPodSpec(tictl.BinlogContainerType, s.Spec.TiDB)
			if oldBinlogImage != newBinlogImage {
				imageUpdate = true
				tictl.SetImageByTiPodSpec(tictl.BinlogContainerType, s.Spec.TiDB, newBinlogImage)
				s.Status.TiDBStatus.UpgradeState = tiapi.Upgrading
				s.Status.TiDBStatus.UpgradeBeginTime = metav1.Now()
			}
		}
	}

	if checkContainerSpec(tictl.TiDBContainerType, c.Tidb) {
		m.revisePodSpec(tictl.TiDBPodType, c.Tidb)
		newTidbImage := tictl.GetImageByTiPodSpec(tictl.TiDBContainerType, c.Tidb)
		oldTidbImage := tictl.GetImageByTiPodSpec(tictl.TiDBContainerType, s.Spec.TiDB)
		if oldTidbImage != newTidbImage {
			imageUpdate = true
			tictl.SetImageByTiPodSpec(tictl.TiDBContainerType, s.Spec.TiDB, newTidbImage)
			s.Status.TiDBStatus.UpgradeState = tiapi.Upgrading
			s.Status.TiDBStatus.UpgradeBeginTime = metav1.Now()
		}
	}

	if imageUpdate {
		_, err = m.Cli.PingcapV1().TidbInstances(namespace).Update(s)
		if err != nil {
			log.Errorf("failed to upgrade TidbInstance %s/%s, update %t, %v", namespace, namespace, imageUpdate, err)
			return imageUpdate, errors.Trace(err)
		}
	}
	return imageUpdate, nil
}

func (m Manager) DeletePod(namespace, podName string) error {
	s := strings.Split(podName, "-")
	if len(s) != 4 { // pod name is in this format tidb-cluster-KIND-xxxxx
		return fmt.Errorf("invalid pod name: %s", podName)
	}
	kind := s[2]
	log.Infof("delete pod %s/%s", namespace, podName)

	pod, err := m.Cli.CoreV1().Pods(namespace).Get(podName, metav1.GetOptions{})
	if err != nil {
		return errors.Trace(err)
	}
	if pod.Annotations == nil {
		pod.Annotations = make(map[string]string)
	}

	switch kind {
	case label.TiDBLabelVal:
		if err := m.Cli.CoreV1().Pods(namespace).Delete(podName, nil); err != nil && !apierrors.IsNotFound(err) {
			log.Errorf("failed to delete TiDB pod %s/%s: %v", namespace, podName, err)
			return err
		}
		log.Infof("TiDB pod %s/%s is deleted", namespace, podName)
		return nil
	case label.PDLabelVal:
		log.Infof("setting Pod: [%s/%s]'s annotations: [%s] to: [%s]", namespace, podName, label.AnnShouldDeletedPD, podName)
		pod.Annotations[label.AnnShouldDeletedPD] = podName
		_, err := m.Cli.CoreV1().Pods(namespace).Update(pod)
		if err != nil {
			log.Errorf("failed to update Pod: [%s/%s], %v", namespace, podName, err)
			return errors.Trace(err)
		}

		log.Infof("update Pod: [%s/%s] successfully", namespace, podName)
		return nil
	case label.TiKVLabelVal:
		log.Infof("setting Pod: [%s/%s]'s annotations: [%s] to: [%s]", namespace, podName, label.AnnShouldDeletedTiKV, podName)
		pod.Annotations[label.AnnShouldDeletedTiKV] = podName
		_, err := m.Cli.CoreV1().Pods(namespace).Update(pod)
		if err != nil {
			log.Errorf("failed to update Pod: [%s/%s], %v", namespace, podName, err)
			return errors.Trace(err)
		}

		log.Infof("update Pod: [%s/%s] successfully", namespace, podName)
		return nil
	default:
		log.Warnf("unknown kind: %s, namespace: %s", kind, namespace)
		return nil
	}
}

func (m Manager) SetPassword(namespace, password string, init bool) (*batchv1.Job, error) {
	cmd := []string{
		"/bin/sh",
		"-c",
		fmt.Sprintf("sleep 5; until mysql -h %s -P4000 -e \"set password for 'root'@'%%'='%s'\" 2>/dev/null; do sleep 1; done", getName("privileged-tidb"), password),
	}
	jobName := getName("set-tidb-password")
	if !init {
		jobName = fmt.Sprintf("%s-%d", getName("reset-tidb-password"), time.Now().Unix())
	}
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:   jobName,
			Labels: label.New().Cluster(clusterName).App("reset-tidb-password").Labels(),
		},
		Spec: batchv1.JobSpec{
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: getName("reset-tidb-password"),
				},
				Spec: v1.PodSpec{
					RestartPolicy: v1.RestartPolicyOnFailure,
					Containers: []v1.Container{
						{
							Name:    "reset-tidb-password",
							Image:   generateImage(tictl.TiDBToolsContainerType, "", ""), // use default tidb-tools
							Command: cmd,
						},
					},
				},
			},
		},
	}
	jobObj, err := m.Cli.BatchV1().Jobs(namespace).Create(job)
	if err != nil {
		log.Errorf("failed to create reset-tidb-password job: %s, namespace: %s, error: %v", jobName, namespace, err)
		return jobObj, err
	}
	return jobObj, nil
}

func (m Manager) GetJobStatus(namespace, jobName string) (JobConditionType, error) {
	job, err := m.Cli.BatchV1().Jobs(namespace).Get(jobName, metav1.GetOptions{})
	if err != nil {
		log.Errorf("failed to get job %s status, namespace: %s, error: %v", jobName, namespace, err)
		return JobUnknown, err
	}
	status := JobRunning
	if job.Status.Succeeded == 1 {
		status = JobComplete
	} else if job.Status.Failed == 1 {
		status = JobFailed
	}
	return status, nil
}

func (m Manager) GetNodes() (*v1.NodeList, error) {
	return m.Cli.CoreV1().Nodes().List(metav1.ListOptions{})
}

func (m Manager) DescribeNode(node *v1.Node) (*Node, error) {
	fieldSelector, err := fields.ParseSelector(
		"spec.nodeName=" + node.Name + ",status.phase!=" + string(v1.PodSucceeded) + ",status.phase!=" + string(v1.PodFailed),
	)
	if err != nil {
		log.Error(err)
		return nil, err
	}

	nodeNonTerminatedPodsList, err := m.Cli.CoreV1().Pods(v1.NamespaceAll).List(metav1.ListOptions{FieldSelector: fieldSelector.String()})
	if err != nil {
		log.Error(err)
		return nil, err
	}

	reqs, limits := getPodsTotalRequestsAndLimits(nodeNonTerminatedPodsList)

	allocatable := node.Status.Capacity
	if len(node.Status.Allocatable) > 0 {
		allocatable = node.Status.Allocatable
	}

	cpuReqs, cpuLimits, memoryReqs, memoryLimits := reqs[v1.ResourceCPU], limits[v1.ResourceCPU], reqs[v1.ResourceMemory], limits[v1.ResourceMemory]
	fractionCpuReqs := float64(cpuReqs.MilliValue()) / float64(allocatable.Cpu().MilliValue()) * 100
	fractionCpuLimits := float64(cpuLimits.MilliValue()) / float64(allocatable.Cpu().MilliValue()) * 100
	fractionMemoryReqs := float64(memoryReqs.Value()) / float64(allocatable.Memory().Value()) * 100
	fractionMemoryLimits := float64(memoryLimits.Value()) / float64(allocatable.Memory().Value()) * 100

	var nodeIP string
	for _, addr := range node.Status.Addresses {
		if addr.Type == v1.NodeInternalIP {
			nodeIP = addr.Address
			break
		}
	}

	return &Node{
		Name:   node.Name,
		IP:     nodeIP,
		Labels: node.Labels,
		CPU: RequestsAndLimits{
			Requests: ValAndFraction{cpuReqs.String(), toFraction(fractionCpuReqs)},
			Limits:   ValAndFraction{cpuLimits.String(), toFraction(fractionCpuLimits)},
		},
		Memory: RequestsAndLimits{
			Requests: ValAndFraction{memoryReqs.String(), toFraction(fractionMemoryReqs)},
			Limits:   ValAndFraction{memoryLimits.String(), toFraction(fractionMemoryLimits)},
		},
	}, nil
}

func (m Manager) Backup(namespace string, param *BackupParam) (*batchv1.Job, error) {
	// mydumper will coredump if using domain name to access tidb
	svc, err := m.Cli.CoreV1().Services(namespace).Get(getName("privileged-tidb"), metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}
	now := time.Now().Unix()
	jobName := fmt.Sprintf("%s-backup-%d", clusterName, now)
	backupName := fmt.Sprintf("%s_%s_%s_%d", namespace, clusterName, param.Database, now)
	threads := param.Threads
	chunks := param.Chunks
	if threads == 0 {
		threads = defaultThreads
	}
	if chunks == 0 {
		chunks = defaultChunks
	}
	fullPath := path.Join(BackupDir, backupName)
	cmds := []string{
		"/mydumper",
		"-h", svc.Spec.ClusterIP,
		"-P", "4000",
		"-u", "root",
		"-t", fmt.Sprintf("%d", threads),
		"-F", fmt.Sprintf("%d", chunks),
		"-B", param.Database,
		"--skip-tz-utc",
		"-o", fullPath,
	}
	tables := strings.Join(param.Tables, ",")
	if len(param.Tables) != 0 {
		cmds = append(cmds, []string{"-T", tables}...)
	}

	pdURLs := []string{fmt.Sprintf("http://%s-pd.%s:2379", clusterName, namespace)}
	pdCli, err := pd.NewClient(pdURLs, pd.SecurityOption{})
	if err != nil {
		errMsg := fmt.Sprintf("failed to create PD client from %v: %v", pdURLs, err)
		log.Error(errMsg)
		return nil, errors.New(errMsg)
	}

	physical, logical, err := pdCli.GetTS(context.Background())
	if err != nil {
		errMsg := fmt.Sprintf("failed to get TS from pd: %v", err)
		log.Error(errMsg)
		return nil, errors.New(errMsg)
	}
	defer pdCli.Close()
	ts := uint64((physical << physicalShiftBits) + logical)

	ann := make(map[string]string)
	ann[AnnBackupDatabase] = param.Database
	ann[AnnBackupTables] = tables
	ann[AnnBackupPath] = fullPath
	ann[AnnBackupSavePoint] = fmt.Sprintf("%d", ts)
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:        jobName,
			Labels:      label.New().Cluster(clusterName).App("backup").Labels(),
			Annotations: ann,
		},
		Spec: batchv1.JobSpec{
			Template: v1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Name: getName("backup"),
				},
				Spec: v1.PodSpec{
					RestartPolicy: v1.RestartPolicyOnFailure,
					NodeSelector:  param.NodeSelector,
					Containers: []v1.Container{
						{
							Name:    "mydumper",
							Image:   generateImage(tictl.TiDBEnterpriseToolsContainerType, "", param.ImageVersion),
							Command: cmds,
							VolumeMounts: []v1.VolumeMount{
								{Name: "data", MountPath: BackupDir},
							},
						},
					},
					Volumes: []v1.Volume{
						{
							Name: "data",
							VolumeSource: v1.VolumeSource{
								HostPath: &v1.HostPathVolumeSource{Path: BackupDir},
							},
						},
					},
				},
			},
		},
	}
	jobObj, err := m.Cli.BatchV1().Jobs(namespace).Create(job)
	if err != nil {
		log.Errorf("failed to create backup job %s/%s: %v", namespace, jobName, err)
		return nil, errors.Trace(err)
	}
	log.Infof("job %s/%s created successfully", namespace, jobName)
	return jobObj, nil
}

func (m Manager) GetBackup(namespace, name string) (*BackupStatus, error) {
	job, err := m.Cli.BatchV1().Jobs(namespace).Get(name, metav1.GetOptions{})
	if err != nil {
		return nil, errors.Trace(err)
	}
	jobStatus := JobRunning
	if job.Status.Succeeded == 1 {
		jobStatus = JobComplete
	} else if job.Status.Failed == 1 {
		jobStatus = JobFailed
	}
	ann := job.Annotations
	savePoint, err := strconv.ParseUint(ann[AnnBackupSavePoint], 10, 64)
	if err != nil {
		log.Errorf("failed to parse savePoint %s in job %s/%s annotation: %v", ann[AnnBackupSavePoint], namespace, name, err)
		return nil, errors.Trace(err)
	}
	status := &BackupStatus{
		Status:    jobStatus,
		Database:  ann[AnnBackupDatabase],
		Tables:    strings.Split(ann[AnnBackupTables], ","),
		Path:      ann[AnnBackupPath],
		SavePoint: savePoint,
	}
	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			map[string]string{
				"job-name": name,
			},
		).String(),
	}
	podList, err := m.Cli.CoreV1().Pods(namespace).List(opts)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(podList.Items) == 0 {
		return nil, errors.New("pod not found")
	}
	pod := podList.Items[0]
	status.Node = pod.Spec.NodeName
	return status, nil
}

func (m Manager) GetVolumes(ns, nodeName string, canSafeDeleted bool) ([]Volume, error) {
	opts := metav1.ListOptions{
		LabelSelector: labels.SelectorFromSet(
			label.New().Cluster(clusterName).Labels(),
		).String(),
	}
	if ns != "" {
		opts = metav1.ListOptions{
			LabelSelector: labels.SelectorFromSet(
				label.New().Namespace(ns).Cluster(clusterName).Labels(),
			).String(),
		}
	}
	pvList, err := m.Cli.CoreV1().PersistentVolumes().List(opts)
	if err != nil {
		log.Errorf("failed to list PV: %v", err)
		return nil, errors.Trace(err)
	}
	var vols []Volume
	for _, pv := range pvList.Items {
		if canSafeDeleted && !pvTombstoned(&pv) && !pvClusterDeleted(&pv) {
			continue
		}

		node := pv.Annotations[volumeprovisioner.AnnProvisionerNode]
		if nodeName != "" && nodeName != node {
			continue
		}
		var capacity string
		if storage, ok := pv.Spec.Capacity[v1.ResourceStorage]; ok {
			capacity = storage.String()
		}
		vol := Volume{
			Name:           pv.Name,
			Namespace:      ns,
			App:            pv.Labels[label.AppLabelKey],
			Node:           pv.Annotations[volumeprovisioner.AnnProvisionerNode],
			HostPath:       pv.Annotations[volumeprovisioner.AnnProvisionerHostPath],
			Capacity:       capacity,
			Used:           pv.Annotations[volumeprovisioner.AnnProvisionerSpaceUsed],
			Pod:            pv.Annotations[label.AnnPodNameKey],
			Cluster:        pv.Labels[label.ClusterLabelKey],
			Status:         string(pv.Status.Phase),
			ClusterID:      pv.Labels[label.ClusterIDLabelKey],
			MemberID:       pv.Labels[label.MemberIDLabelKey],
			StoreID:        pv.Labels[label.StoreIDLabelKey],
			Tombstoned:     pvTombstoned(&pv),
			ClusterDeleted: pvClusterDeleted(&pv),
		}
		vols = append(vols, vol)
	}
	return vols, nil
}

func (m Manager) DeleteVolume(volName string) error {
	pv, err := m.Cli.CoreV1().PersistentVolumes().Get(volName, metav1.GetOptions{})
	if err != nil {
		return err
	}

	if !(pvTombstoned(pv) || pvClusterDeleted(pv)) {
		return errors.Errorf("PV: [%s] is not tombstoned and clusterDeleted", pv.GetName())
	}

	t := time.Now().Format(time.RFC3339)
	pv.Annotations[volumeprovisioner.AnnPVManualDeleted] = t

	_, err = m.Cli.CoreV1().PersistentVolumes().Update(pv)
	if err != nil {
		log.Errorf("failed to update PV [%s], %v", volName, err)
		return err
	}

	log.Infof("update PV [%s]'s Annotations[%s] to [%s] successfully",
		volName, volumeprovisioner.AnnPVManualDeleted, t)
	return nil
}

func pvTombstoned(pv *v1.PersistentVolume) bool {
	return pv.Annotations[volumeprovisioner.AnnPVTombstoned] != ""
}

func pvClusterDeleted(pv *v1.PersistentVolume) bool {
	return pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] != ""
}