test_delete_backup.go

package sharded

import (
	"fmt"
	"io/ioutil"
	"log"
	"net/url"
	"os"
	"strings"
	"time"

	"github.com/minio/minio-go"
	"github.com/percona/percona-backup-mongodb/pbm"
	"gopkg.in/yaml.v2"
)

type backupDelete struct {
	name string
	ts   time.Time
}

func (c *Cluster) BackupDelete(storage string) {
	c.pitrOn()
	defer c.pitrOff()

	checkData := c.DataChecker()

	backups := make([]backupDelete, 5)
	for i := 0; i < len(backups); i++ {
		ts := time.Now()
		time.Sleep(1 * time.Second)
		c.printBcpList()
		bcpName := c.LogicalBackup()
		backups[i] = backupDelete{
			name: bcpName,
			ts:   ts,
		}
		c.BackupWaitDone(bcpName)

		time.Sleep(time.Second * 60)
	}

	c.printBcpList()

	log.Println("delete backup", backups[4].name)
	_, err := c.pbm.RunCmd("pbm", "delete-backup", "-f", backups[4].name)
	if err != nil {
		log.Fatalf("ERROR: delete backup %s: %v", backups[4].name, err)
	}
	log.Println("wait for delete")
	err = c.mongopbm.WaitConcurentOp(&pbm.LockHeader{Type: pbm.CmdDeleteBackup}, time.Minute*5)
	if err != nil {
		log.Fatalf("waiting for the delete: %v", err)
	}

	c.printBcpList()

	log.Printf("delete backups older than %s / %s \n", backups[3].name, backups[3].ts.Format("2006-01-02T15:04:05"))
	_, err = c.pbm.RunCmd("pbm", "delete-backup", "-f", "--older-than", backups[3].ts.Format("2006-01-02T15:04:05"))
	if err != nil {
		log.Fatalf("ERROR: delete backups older than %s: %v", backups[3].name, err)
	}
	log.Println("wait for delete")
	err = c.mongopbm.WaitConcurentOp(&pbm.LockHeader{Type: pbm.CmdDeleteBackup}, time.Minute*5)
	if err != nil {
		log.Fatalf("waiting for the delete: %v", err)
	}
	time.Sleep(time.Second * 5)
	c.pitrOff()

	c.printBcpList()

	left := map[string]struct{}{
		backups[0].name: {}, // is a base for the pitr timeline, shouldn't be deleted
		backups[3].name: {},
	}
	log.Println("should be only backups", left)
	checkArtefacts(storage, left)

	blist, err := c.mongopbm.BackupsList(0)
	if err != nil {
		log.Fatalln("ERROR: get backups list", err)
	}

	if len(blist) != len(left) {
		log.Fatalf("ERROR: backups list mismatch, expect: %v, have: %v", left, blist)
	}
	for _, b := range blist {
		if _, ok := left[b.Name]; !ok {
			log.Fatalf("ERROR: backup %s should be deleted", b.Name)
		}
	}

	list, err := c.pbm.List()
	if err != nil {
		log.Fatalf("ERROR: get backups/pitr list: %v", err)
	}

	if len(list.Snapshots) == 0 {
		log.Println("ERROR: empty spanpshots list")
		c.printBcpList()
		os.Exit(1)
	}

	tsp := time.Unix(list.Snapshots[len(list.Snapshots)-1].StateTS, 0).Add(time.Second * 10)

	log.Printf("delete pitr older than %s \n", tsp.Format("2006-01-02T15:04:05"))
	_, err = c.pbm.RunCmd("pbm", "delete-pitr", "-f", "--older-than", tsp.Format("2006-01-02T15:04:05"))
	if err != nil {
		log.Fatalf("ERROR: delete pitr older than %s: %v", tsp.Format("2006-01-02T15:04:05"), err)
	}
	log.Println("wait for delete-pitr")
	err = c.mongopbm.WaitConcurentOp(&pbm.LockHeader{Type: pbm.CmdDeletePITR}, time.Minute*5)
	if err != nil {
		log.Fatalf("ERROR: waiting for the delete-pitr: %v", err)
	}

	list, err = c.pbm.List()
	if err != nil {
		log.Fatalf("ERROR: get backups/pitr list: %v", err)
	}

	if len(list.PITR.Ranges) == 0 {
		log.Fatalf("ERROR: empty pitr list, expected range after the last backup")
	}
	if len(list.Snapshots) == 0 {
		log.Fatalf("ERROR: empty spanpshots list")
	}

	if list.PITR.Ranges[0].Range.Start-1 != uint32(list.Snapshots[len(list.Snapshots)-1].StateTS) {
		log.Printf("ERROR: expected range starts the next second after the backup complete time, %v | %v", list.PITR.Ranges[0].Range.Start+1, uint32(list.Snapshots[len(list.Snapshots)-1].StateTS))
		c.printBcpList()
		os.Exit(1)
	}

	log.Println("delete pitr all")
	_, err = c.pbm.RunCmd("pbm", "delete-pitr", "-f", "--all")
	if err != nil {
		log.Fatalf("ERROR: delete all pitr")
	}
	log.Println("wait for delete-pitr all")
	err = c.mongopbm.WaitConcurentOp(&pbm.LockHeader{Type: pbm.CmdDeletePITR}, time.Minute*5)
	if err != nil {
		log.Fatalf("ERROR: waiting for the delete-pitr: %v", err)
	}

	list, err = c.pbm.List()
	if err != nil {
		log.Fatalf("ERROR: get backups/pitr list: %v", err)
	}

	if len(list.PITR.Ranges) != 0 {
		log.Println("ERROR: expected empty pitr list but got:")
		c.printBcpList()
		os.Exit(1)
	}

	log.Println("trying to restore from", backups[3])
	c.DeleteBallast()
	c.LogicalRestore(backups[3].name)
	checkData()
}

const awsurl = "s3.amazonaws.com"

// checkArtefacts checks if all backups artifacts removed
// except for the shouldStay
func checkArtefacts(conf string, shouldStay map[string]struct{}) {
	log.Println("check all artifacts deleted excepts backup's", shouldStay)
	buf, err := ioutil.ReadFile(conf)
	if err != nil {
		log.Fatalln("ERROR: unable to read config file:", err)
	}

	var cfg pbm.Config
	err = yaml.UnmarshalStrict(buf, &cfg)
	if err != nil {
		log.Fatalln("ERROR: unmarshal yaml:", err)
	}

	stg := cfg.Storage

	endopintURL := awsurl
	if stg.S3.EndpointURL != "" {
		eu, err := url.Parse(stg.S3.EndpointURL)
		if err != nil {
			log.Fatalln("ERROR: parse EndpointURL:", err)
		}
		endopintURL = eu.Host
	}

	mc, err := minio.NewWithRegion(endopintURL, stg.S3.Credentials.AccessKeyID, stg.S3.Credentials.SecretAccessKey, false, stg.S3.Region)
	if err != nil {
		log.Fatalln("ERROR: NewWithRegion:", err)
	}

	for object := range mc.ListObjects(stg.S3.Bucket, stg.S3.Prefix, true, nil) {
		if strings.Contains(object.Key, pbm.StorInitFile) || strings.Contains(object.Key, "/pbmPitr/") {
			continue
		}
		if object.Err != nil {
			fmt.Println("ERROR: ListObjects: ", object.Err)
			continue
		}

		var ok bool
		for b := range shouldStay {
			if strings.Contains(object.Key, b) {
				ok = true
				break
			}
		}
		if !ok {
			log.Fatalln("ERROR: failed to delete lefover", object.Key)
		}
	}
}

func (c *Cluster) BackupNotDeleteRunning() {
	bcpName := c.LogicalBackup()
	c.printBcpList()
	log.Println("deleting backup", bcpName)
	o, err := c.pbm.RunCmd("pbm", "delete-backup", "-f", bcpName)
	if err == nil || !strings.Contains(err.Error(), "unable to delete backup in running state") {
		list, lerr := c.pbm.RunCmd("pbm", "list")
		log.Fatalf("ERROR: running backup '%s' shouldn't be deleted.\nOutput: %s\nStderr:%v\nBackups list:\n%v\n%v", bcpName, o, err, list, lerr)
	}
	c.BackupWaitDone(bcpName)
	time.Sleep(time.Second * 2)
}

func (c *Cluster) printBcpList() {
	listo, _ := c.pbm.RunCmd("pbm", "list", "--full")
	fmt.Printf("backup list:\n%s\n", listo)
}