add schduler test case

tests/actions.go

@@ -31,8 +31,10 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/label"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 )
@@ -48,6 +50,11 @@ func NewOperatorActions(cli versioned.Interface, kubeCli kubernetes.Interface) O
 const (
 	DefaultPollTimeout  time.Duration = 10 * time.Minute
 	DefaultPollInterval time.Duration = 1 * time.Minute
+	getBackupDirPodName               = "get-backup-dir"
+)
+
+var (
+	backupDir string
 )
 
 type OperatorActions interface {
@@ -74,6 +81,7 @@ type OperatorActions interface {
 	CleanMonitor(info *TidbClusterInfo) error
 	ForceDeploy(info *TidbClusterInfo) error
 	CreateSecret(info *TidbClusterInfo) error
+	getBackupDir(info *TidbClusterInfo) (string, error)
 }
 
 type FaultTriggerActions interface {
@@ -113,6 +121,7 @@ type OperatorInfo struct {
 }
 
 type TidbClusterInfo struct {
+	Name             string
 	Namespace        string
 	ClusterName      string
 	OperatorTag      string
@@ -121,6 +130,7 @@ type TidbClusterInfo struct {
 	TiDBImage        string
 	StorageClassName string
 	Password         string
+	InitSql          string
 	RecordCount      string
 	InsertBetchSize  string
 	Resources        map[string]string
@@ -129,9 +139,6 @@ type TidbClusterInfo struct {
 
 func (tc *TidbClusterInfo) HelmSetString() string {
 
-	// add a database and table for test
-	initSql := `"create database record;use record;create table test(t char(32));"`
-
 	set := map[string]string{
 		"clusterName":             tc.ClusterName,
 		"pd.storageClassName":     tc.StorageClassName,
@@ -143,7 +150,7 @@ func (tc *TidbClusterInfo) HelmSetString() string {
 		"tikv.image":              tc.TiKVImage,
 		"tidb.image":              tc.TiDBImage,
 		"tidb.passwordSecretName": "set-secret",
-		"tidb.initSql":            initSql,
+		"tidb.initSql":            tc.InitSql,
 	}
 
 	for k, v := range tc.Resources {
@@ -243,6 +250,7 @@ func (oa *operatorActions) CleanTidbCluster(info *TidbClusterInfo) error {
 		info.ClusterName,
 		fmt.Sprintf("%s-backup", info.ClusterName),
 		fmt.Sprintf("%s-restore", info.ClusterName),
+		fmt.Sprintf("%s-scheduler-backup", info.ClusterName),
 	}
 	for _, chartName := range charts {
 		res, err := exec.Command("helm", "del", "--purge", chartName).CombinedOutput()
@@ -252,6 +260,11 @@ func (oa *operatorActions) CleanTidbCluster(info *TidbClusterInfo) error {
 		}
 	}
 
+	_, err := oa.kubeCli.CoreV1().Pods(info.Namespace).Get(getBackupDirPodName, metav1.GetOptions{})
+	if !errors.IsNotFound(err) {
+		oa.kubeCli.CoreV1().Pods(info.Namespace).Delete(getBackupDirPodName, &metav1.DeleteOptions{})
+	}
+
 	setStr := label.New().Instance(info.ClusterName).String()
 
 	resources := []string{"pvc"}
@@ -847,9 +860,11 @@ func (oa *operatorActions) DeployAdHocBackup(info *TidbClusterInfo) error {
 	defer func() {
 		glog.Infof("deploy adhoc backup end cluster[%s] namespace[%s]", info.ClusterName, info.Namespace)
 	}()
+	// I need a name for pvc and another name for backupDir,
+	// But in operator, there are the same
 	sets := map[string]string{
 		"clusterName":  info.ClusterName,
-		"name":         "test-backup",
+		"name":         info.Name,
 		"mode":         "backup",
 		"user":         "root",
 		"password":     info.Password,
@@ -873,6 +888,7 @@ func (oa *operatorActions) DeployAdHocBackup(info *TidbClusterInfo) error {
 	if err != nil {
 		return fmt.Errorf("failed to launch adhoc backup job: %v, %s", err, string(res))
 	}
+
 	return nil
 }
 
@@ -882,7 +898,7 @@ func (oa *operatorActions) CheckAdHocBackup(info *TidbClusterInfo) error {
 		glog.Infof("deploy clean backup end cluster[%s] namespace[%s]", info.ClusterName, info.Namespace)
 	}()
 
-	jobName := fmt.Sprintf("%s-%s", info.ClusterName, "test-backup")
+	jobName := fmt.Sprintf("%s-%s", info.ClusterName, info.Name)
 	fn := func() (bool, error) {
 		job, err := oa.kubeCli.BatchV1().Jobs(info.Namespace).Get(jobName, metav1.GetOptions{})
 		if err != nil {
@@ -901,6 +917,12 @@ func (oa *operatorActions) CheckAdHocBackup(info *TidbClusterInfo) error {
 	if err != nil {
 		return fmt.Errorf("failed to launch scheduler backup job: %v", err)
 	}
+
+	backupDir, err = oa.getBackupDir(info)
+	if err != nil {
+		return fmt.Errorf("failed to get backup dir: %v", err)
+	}
+
 	return nil
 }
 
@@ -911,7 +933,7 @@ func (oa *operatorActions) Restore(from *TidbClusterInfo, to *TidbClusterInfo) e
 	}()
 	sets := map[string]string{
 		"clusterName":  to.ClusterName,
-		"name":         "test-backup",
+		"name":         to.Name,
 		"mode":         "restore",
 		"user":         "root",
 		"password":     to.Password,
@@ -945,7 +967,7 @@ func (oa *operatorActions) CheckRestore(from *TidbClusterInfo, to *TidbClusterIn
 		glog.Infof("check restore end cluster[%s] namespace[%s]", to.ClusterName, to.Namespace)
 	}()
 
-	jobName := fmt.Sprintf("%s-restore-test-backup", to.ClusterName)
+	jobName := fmt.Sprintf("%s-restore-%s", to.ClusterName, from.Name)
 	fn := func() (bool, error) {
 		job, err := oa.kubeCli.BatchV1().Jobs(to.Namespace).Get(jobName, metav1.GetOptions{})
 		if err != nil {
@@ -1065,13 +1087,196 @@ func releaseIsExist(err error) bool {
 }
 
 func (oa *operatorActions) DeployScheduledBackup(info *TidbClusterInfo) error {
+	glog.Infof("begin to deploy scheduled backup")
+	defer func() {
+		glog.Infof("deploy shceduled backup end")
+	}()
+	minute := time.Now().Minute()
+	cron := fmt.Sprintf("'%d * * * *'", (minute+5)%60)
+	sets := map[string]string{
+		"clusterName":              info.ClusterName,
+		"scheduledBackup.create":   "true",
+		"scheduledBackup.user":     "root",
+		"scheduledBackup.password": info.Password,
+		"scheduledBackup.schedule": cron,
+		"scheduledBackup.storage":  "10Gi",
+	}
+	var buffer bytes.Buffer
+	for k, v := range sets {
+		set := fmt.Sprintf(" --set %s=%s", k, v)
+		_, err := buffer.WriteString(set)
+		if err != nil {
+			return err
+		}
+	}
+
+	setStr := buffer.String()
+
+	cmd := fmt.Sprintf("helm upgrade %s /charts/%s/tidb-cluster %s",
+		info.ClusterName, info.OperatorTag, setStr)
+
+	glog.Infof("scheduled-backup delploy [%s]", cmd)
+	res, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("failed to launch scheduler backup job: %v, %s", err, string(res))
+	}
 	return nil
 }
 
 func (oa *operatorActions) CheckScheduledBackup(info *TidbClusterInfo) error {
+	glog.Infof("begin to check scheduler backup cluster[%s] namespace[%s]", info.ClusterName, info.Namespace)
+	defer func() {
+		glog.Infof("deploy check scheduler end cluster[%s] namespace[%s]", info.ClusterName, info.Namespace)
+	}()
+
+	jobName := fmt.Sprintf("%s-scheduled-backup", info.ClusterName)
+	fn := func() (bool, error) {
+		job, err := oa.kubeCli.BatchV1beta1().CronJobs(info.Namespace).Get(jobName, metav1.GetOptions{})
+		if err != nil {
+			glog.Errorf("failed to get cronjobs %s ,%v", jobName, err)
+			return false, nil
+		}
+
+		jobs, err := oa.kubeCli.BatchV1().Jobs(info.Namespace).List(metav1.ListOptions{})
+		if err != nil {
+			glog.Errorf("failed to list jobs %s ,%v", info.Namespace, err)
+			return false, nil
+		}
+
+		backupJobs := []batchv1.Job{}
+		for _, j := range jobs.Items {
+			if pid, found := getParentUIDFromJob(j); found && pid == job.UID {
+				backupJobs = append(backupJobs, j)
+			}
+		}
+
+		if len(backupJobs) == 0 {
+			glog.Errorf("cluster [%s] scheduler jobs is creating, please wait!", info.ClusterName)
+			return false, nil
+		}
+
+		for _, j := range backupJobs {
+			if j.Status.Succeeded == 0 {
+				glog.Errorf("cluster [%s] back up job is not completed, please wait! ", info.ClusterName)
+				return false, nil
+			}
+		}
+
+		return true, nil
+	}
+
+	err := wait.Poll(DefaultPollInterval, DefaultPollTimeout, fn)
+	if err != nil {
+		return fmt.Errorf("failed to launch scheduler backup job: %v", err)
+	}
+
+	backupDir, err = oa.getBackupDir(info)
+	if err != nil {
+		return fmt.Errorf("failed to get backup dir: %v", err)
+	}
+
 	return nil
 }
 
+func getParentUIDFromJob(j batchv1.Job) (types.UID, bool) {
+	controllerRef := metav1.GetControllerOf(&j)
+
+	if controllerRef == nil {
+		return types.UID(""), false
+	}
+
+	if controllerRef.Kind != "CronJob" {
+		glog.Infof("Job with non-CronJob parent, name %s namespace %s", j.Name, j.Namespace)
+		return types.UID(""), false
+	}
+
+	return controllerRef.UID, true
+}
+
+func (oa *operatorActions) getBackupDir(info *TidbClusterInfo) (string, error) {
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      getBackupDirPodName,
+			Namespace: info.Namespace,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:    getBackupDirPodName,
+					Image:   "pingcap/tidb-cloud-backup:latest",
+					Command: []string{"sleep", "3000"},
+					VolumeMounts: []corev1.VolumeMount{
+						{
+							Name:      "data",
+							MountPath: "/data",
+						},
+					},
+				},
+			},
+			Volumes: []corev1.Volume{
+				{
+					Name: "data",
+					VolumeSource: corev1.VolumeSource{
+						PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+							ClaimName: info.Name,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	fn := func() (bool, error) {
+		_, err := oa.kubeCli.CoreV1().Pods(info.Namespace).Get(getBackupDirPodName, metav1.GetOptions{})
+		if !errors.IsNotFound(err) {
+			return false, nil
+		}
+		return true, nil
+	}
+
+	err := wait.Poll(DefaultPollInterval, DefaultPollTimeout, fn)
+
+	if err != nil {
+		return "", fmt.Errorf("failed to delete pod %s", getBackupDirPodName)
+	}
+
+	_, err = oa.kubeCli.CoreV1().Pods(info.Namespace).Create(pod)
+	if err != nil && !errors.IsAlreadyExists(err) {
+		glog.Errorf("cluster: [%s/%s] create get backup dir pod failed, error :%v", info.Namespace, info.ClusterName, err)
+		return "", err
+	}
+
+	fn = func() (bool, error) {
+		_, err := oa.kubeCli.CoreV1().Pods(info.Namespace).Get(getBackupDirPodName, metav1.GetOptions{})
+		if errors.IsNotFound(err) {
+			return false, nil
+		}
+		return true, nil
+	}
+
+	err = wait.Poll(DefaultPollInterval, DefaultPollTimeout, fn)
+
+	if err != nil {
+		return "", fmt.Errorf("failed to create pod %s", getBackupDirPodName)
+	}
+
+	cmd := fmt.Sprintf("kubectl exec %s -n %s ls /data", getBackupDirPodName, info.Namespace)
+	glog.Infof(cmd)
+	res, err := exec.Command("/bin/sh", "-c", cmd).CombinedOutput()
+	if err != nil {
+		glog.Errorf("cluster:[%s/%s] exec :%s failed,error:%v,result:%s", info.Namespace, info.ClusterName, cmd, err, res)
+		return "", err
+	}
+
+	dirs := strings.Split(string(res), "\n")
+	glog.Infof("dirs in pod info name [%s] dir name [%s]", info.Name, strings.Join(dirs, ","))
+	return strings.TrimSpace(dirs[0]), nil
+}
+
+func (info *TidbClusterInfo) FullName() string {
+	return fmt.Sprintf("%s/%s", info.Namespace, info.ClusterName)
+}
+
 func (oa *operatorActions) DeployIncrementalBackup(from *TidbClusterInfo, to *TidbClusterInfo) error {
 	return nil
 }

tests/backup/backupcase.go

@@ -70,5 +70,20 @@ func (bc *BackupCase) Run() error {
 		return err
 	}
 
+	bc.srcCluster.Name = "demo-scheduled-backup"
+	bc.desCluster.Name = "demo-scheduled-backup"
+
+	err = bc.operator.DeployScheduledBackup(bc.srcCluster)
+	if err != nil {
+		glog.Errorf("cluster:[%s] scheduler happen error: %v", bc.srcCluster.ClusterName, err)
+		return err
+	}
+
+	err = bc.operator.CheckScheduledBackup(bc.srcCluster)
+	if err != nil {
+		glog.Errorf("cluster:[%s] scheduler failed error: %v", bc.srcCluster.ClusterName, err)
+		return err
+	}
+
 	return nil
 }