velero/pkg/uploader/block/uploader.go

/*
Copyright The Velero Contributors.

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,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package block

import (
	"context"
	"io"
	"os"
	"runtime"
	"strings"

	"github.com/cockroachdb/errors"
	"github.com/sirupsen/logrus"

	"github.com/vmware-tanzu/velero/pkg/repository/udmrepo"
	"github.com/vmware-tanzu/velero/pkg/uploader"
	cbt "github.com/vmware-tanzu/velero/pkg/uploader/cbt/types"
	"github.com/vmware-tanzu/velero/pkg/util/freelist"
)

var ErrCanceled = errors.New("uploader is canceled")

const (
	blockSize  = (1 << 20)
	bufferSize = 100 << 20
)

type sourceInfo struct {
	dev        *os.File
	realSource string
	size       int64
}

type destInfo struct {
	dev  *os.File
	path string
}

type Uploader interface {
	Backup(sourceInfo, udmrepo.ID, cbt.Iterator, map[string]string) (udmrepo.Snapshot, int64, error)
	Restore(udmrepo.Snapshot, destInfo, cbt.Iterator, map[string]string) (int64, error)
}

type blockUploader struct {
	ctx        context.Context
	repoWriter udmrepo.BackupRepo
	progress   uploader.ProgressUpdater
	log        logrus.FieldLogger
}

func NewUploader(ctx context.Context, repoWriter udmrepo.BackupRepo, progress uploader.ProgressUpdater, log logrus.FieldLogger) Uploader {
	return &blockUploader{
		ctx:        ctx,
		repoWriter: repoWriter,
		progress:   progress,
		log:        log,
	}
}

func (blkup *blockUploader) Backup(source sourceInfo, parentObject udmrepo.ID, bitmap cbt.Iterator, configs map[string]string) (udmrepo.Snapshot, int64, error) {
	snapStart := blkup.repoWriter.Time()

	if bitmap == nil {
		return udmrepo.Snapshot{}, 0, errors.New("bitmap is not available")
	}

	backupMode := udmrepo.ObjectDataBackupModeInc
	if parentObject == "" {
		backupMode = udmrepo.ObjectDataBackupModeFull
	}

	destObj, err := blkup.repoWriter.NewObjectWriter(blkup.ctx, udmrepo.ObjectWriteOptions{
		Description:  "BDEV:" + getObjectName(source.realSource),
		DataType:     udmrepo.ObjectDataTypeData,
		AccessMode:   udmrepo.ObjectDataAccessModeBlock,
		ParentObject: parentObject,
		BackupMode:   backupMode,
		AsyncWrites:  runtime.NumCPU(),
	})
	if err != nil {
		return udmrepo.Snapshot{}, 0, errors.Wrap(err, "error creating object writer")
	}

	defer destObj.Close()

	id, backupSize, objectSize, err := blkup.backupObject(source.dev, destObj, bitmap, source.size)
	if err != nil {
		return udmrepo.Snapshot{}, 0, errors.Wrapf(err, "error backing up bdev %s", source.realSource)
	}

	entryID, err := blkup.repoWriter.WriteMetadata(blkup.ctx, &udmrepo.Metadata{
		SubObjects: []udmrepo.ObjectMetadata{
			{
				ID:          id,
				Name:        getObjectName(source.realSource),
				Type:        udmrepo.ObjectDataTypeData,
				Size:        objectSize,
				Permissions: 0o777,
			},
		},
	},
		udmrepo.ObjectWriteOptions{
			Description: "bdev-root",
		})
	if err != nil {
		return udmrepo.Snapshot{}, 0, errors.Wrap(err, "error writing metadata")
	}

	snapEnd := blkup.repoWriter.Time()

	return udmrepo.Snapshot{
		Source:      source.realSource,
		StartTime:   snapStart,
		EndTime:     snapEnd,
		Description: source.realSource,
		TotalSize:   objectSize,
		RootObject: udmrepo.ObjectMetadata{
			ID:          entryID,
			Name:        "bdev-root",
			Type:        udmrepo.ObjectDataTypeMetadata,
			Permissions: 0o777,
		},
	}, backupSize, nil
}

// TODO implement in following PRs
func (blkup *blockUploader) Restore(snapshot udmrepo.Snapshot, dest destInfo, bitmap cbt.Iterator, configs map[string]string) (int64, error) {
	return 0, errors.New("not implemented")
}

func (blkup *blockUploader) backupObject(dev *os.File, dest udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (udmrepo.ID, int64, int64, error) {
	backupSize, objectSize, err := blkup.backupData(dev, dest, bitmap, totalLength)
	if err != nil {
		return "", backupSize, objectSize, err
	}

	id, err := dest.Result()
	return id, backupSize, objectSize, err
}

type readResult struct {
	buffer []byte
	offset int64
	err    error
}

func (r *readResult) resetBuffer(list *freelist.FreeList) {
	if r.buffer != nil {
		list.Return(r.buffer)
		r.buffer = nil
	}
}

func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (int64, int64, error) {
	blockSize := bitmap.BlockSize()
	list := freelist.New(bufferSize, int(blockSize))
	resultChan := make(chan readResult, list.Capacity())
	totalCount := bitmap.Count()
	aligned := (totalLength + int64(blockSize) - 1) / int64(blockSize) * int64(blockSize)

	quit := make(chan struct{})
	defer close(quit)

	go func() {
		defer close(resultChan)

		offset, valid := bitmap.Next()
		var buffer []byte
		for valid {
			select {
			case <-blkup.ctx.Done():
				return
			case <-quit:
				return
			case buffer = <-list.Chunks():
			}

			length := blockSize
			if offset+uint64(length) > uint64(totalLength) {
				length = uint(uint64(totalLength) - offset)
				clear(buffer)
			}

			readBytes, err := reader.ReadAt(buffer[:length], int64(offset))
			if err == nil && readBytes <= 0 {
				err = io.ErrUnexpectedEOF
			}

			r := readResult{
				buffer: buffer,
				offset: int64(offset),
				err:    err,
			}

			if r.err != nil {
				r.resetBuffer(list)
			}

			resultChan <- r

			if r.err != nil {
				return
			}

			offset, valid = bitmap.Next()
		}
	}()

	var lastPos int64
	var result readResult
	var written int64
	var curCount int64
	var writeErr error
	var readerRunning bool

	for curCount < int64(totalCount) {
		select {
		case <-blkup.ctx.Done():
			writeErr = ErrCanceled
		case result, readerRunning = <-resultChan:
			if !readerRunning {
				if blkup.ctx.Err() != nil {
					writeErr = ErrCanceled
				} else {
					writeErr = io.ErrUnexpectedEOF
				}
			}
		}

		if writeErr != nil {
			break
		}

		if result.err != nil {
			writeErr = result.err
			break
		}

		n, err := writer.WriteAt(result.buffer, result.offset)
		if err != nil {
			writeErr = err
			break
		}

		if blockSize != uint(n) {
			writeErr = io.ErrShortWrite
			break
		}

		written += int64(blockSize)
		lastPos = result.offset + int64(blockSize)
		result.resetBuffer(list)
		curCount++

		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: lastPos, TotalBytes: aligned})
	}

	result.resetBuffer(list)

	if writeErr != nil {
		return written, aligned, writeErr
	}

	if lastPos < aligned {
		s, err := copyTailData(reader, writer, totalLength, int64(blockSize))
		if err != nil {
			return written, aligned, errors.Wrapf(err, "unable to write tail data at %v", lastPos)
		}

		written += s

		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: aligned, TotalBytes: aligned})
	}

	return written, aligned, nil
}

func copyTailData(source io.ReaderAt, writer udmrepo.ObjectWriter, totalLength int64, blockSize int64) (int64, error) {
	roundUp := (totalLength + blockSize - 1) / blockSize * blockSize
	roundDown := totalLength / blockSize * blockSize
	length := totalLength - roundDown

	if length == 0 {
		if _, err := writer.WriteAt(nil, roundUp); err != nil {
			return -1, errors.Wrapf(err, "error writing sparse to %v", roundUp)
		}
	} else {
		buffer := make([]byte, blockSize)
		if _, err := source.ReadAt(buffer[:length], roundDown); err != nil {
			return -1, errors.Wrapf(err, "error reading tail data with length %v", length)
		}

		if _, err := writer.WriteAt(buffer, roundDown); err != nil {
			return -1, errors.Wrapf(err, "error writing tail data at %v", roundDown)
		}
	}

	return length, nil
}

func getObjectName(source string) string {
	s := strings.ReplaceAll(source, "/", "-")
	s = strings.ReplaceAll(s, "\\", "-")
	return strings.Trim(s, "-")
}

func loadObjectFromSnapshot(ctx context.Context, rep udmrepo.BackupRepo, snapshot *udmrepo.Snapshot) (udmrepo.ID, error) {
	if snapshot == nil {
		return "", errors.New("snapshot is empty")
	}

	meta, err := rep.ReadMetadata(ctx, snapshot.RootObject.ID)
	if err != nil {
		return "", errors.Wrapf(err, "error reading snapshot metadata for %s", snapshot.Description)
	}

	if len(meta.SubObjects) != 1 {
		return "", errors.Errorf("unexpected number of bdev object (%d) for snapshot %s", len(meta.SubObjects), snapshot.Description)
	}

	return meta.SubObjects[0].ID, nil
}