Implemented tar+zstd compression + unit tests for it.
This commit is contained in:
@@ -142,8 +142,8 @@ func LoadConfig(path string) error {
|
||||
log.Warnf("Config option compression_algorithm is not set, fallback to %s", C.CompressionAlgorithm)
|
||||
}
|
||||
|
||||
if C.CompressionAlgorithm != "gzip" && C.CompressionAlgorithm != "bzip2" {
|
||||
return fmt.Errorf("Compression_algorithm must be gzip or bzip2, got %s", C.CompressionAlgorithm)
|
||||
if C.CompressionAlgorithm != "gzip" && C.CompressionAlgorithm != "bzip2" && C.CompressionAlgorithm != "zstd" {
|
||||
return fmt.Errorf("Compression_algorithm must be gzip, bzip2 or zstd, got %s", C.CompressionAlgorithm)
|
||||
}
|
||||
|
||||
// Compile exclude patterns.
|
||||
|
||||
@@ -253,6 +253,27 @@ func TestLoadConfigCompressionAlgorithmBzip2(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadConfigCompressionAlgorithmZstd tests zstd compression algorithm.
|
||||
func TestLoadConfigCompressionAlgorithmZstd(t *testing.T) {
|
||||
config := `{
|
||||
"user": "test",
|
||||
"password": "test",
|
||||
"directories": ["../../test_data/test1/foo"],
|
||||
"nohttps": true,
|
||||
"compression_algorithm": "zstd"
|
||||
}`
|
||||
tmpFile := writeTempConfig(t, config)
|
||||
|
||||
err := LoadConfig(tmpFile)
|
||||
if err != nil {
|
||||
t.Errorf("Expected no error, got %v", err)
|
||||
}
|
||||
|
||||
if C.CompressionAlgorithm != "zstd" {
|
||||
t.Errorf("C.CompressionAlgorithm should be zstd, got %s", C.CompressionAlgorithm)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadConfigCompressionAlgorithmInvalid tests invalid compression algorithm.
|
||||
func TestLoadConfigCompressionAlgorithmInvalid(t *testing.T) {
|
||||
config := `{
|
||||
|
||||
@@ -78,10 +78,16 @@ func NewServer(configPath string) (*http.Server, error) {
|
||||
extension string
|
||||
)
|
||||
|
||||
if C.CompressionAlgorithm == "bzip2" {
|
||||
switch C.CompressionAlgorithm {
|
||||
case "bzip2":
|
||||
archive = CreateTarBzip2Stream(ctx, files)
|
||||
extension = "tar.bz2"
|
||||
} else {
|
||||
|
||||
case "zstd":
|
||||
archive = CreateTarZstdStream(ctx, files)
|
||||
extension = "tar.zst"
|
||||
|
||||
default:
|
||||
archive = CreateTarGzStream(ctx, files)
|
||||
extension = "tar.gz"
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"backer/internal/log"
|
||||
|
||||
"github.com/dsnet/compress/bzip2"
|
||||
"github.com/klauspost/compress/zstd"
|
||||
)
|
||||
|
||||
// CreateTarGzStream takes a list of files and returns a reader that reads from a tar archive containing these files.
|
||||
@@ -303,6 +304,157 @@ func CreateTarBzip2Stream(ctx context.Context, filepaths []string) io.ReadCloser
|
||||
return pr
|
||||
}
|
||||
|
||||
// zstdLevel returns the zstd encoder level based on the 1-9 compression level.
|
||||
func zstdLevel(level int) zstd.EncoderLevel {
|
||||
switch {
|
||||
case level <= 3:
|
||||
return zstd.SpeedFastest
|
||||
case level <= 6:
|
||||
return zstd.SpeedDefault
|
||||
default:
|
||||
return zstd.SpeedBetterCompression
|
||||
}
|
||||
}
|
||||
|
||||
// CreateTarZstdStream takes a list of files and returns a reader that reads from a zstd-compressed tar archive.
|
||||
func CreateTarZstdStream(ctx context.Context, filepaths []string) io.ReadCloser {
|
||||
pr, pw := io.Pipe()
|
||||
|
||||
go func() {
|
||||
var pipeErr error
|
||||
|
||||
closePipeWithError := func(err error) {
|
||||
if pipeErr != nil {
|
||||
return
|
||||
}
|
||||
|
||||
pipeErr = err
|
||||
pw.CloseWithError(err)
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if pipeErr == nil {
|
||||
pw.Close()
|
||||
}
|
||||
}()
|
||||
|
||||
zw, err := zstd.NewWriter(pw, zstd.WithEncoderLevel(zstdLevel(C.CompressionLevel)))
|
||||
if err != nil {
|
||||
closePipeWithError(fmt.Errorf("zstd writer error: %w", err))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
defer func() {
|
||||
if pipeErr == nil {
|
||||
if err := zw.Close(); err != nil {
|
||||
if !isPipeClosedError(err) {
|
||||
log.Errorf("Failed to close zstd writer: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
tw := tar.NewWriter(zw)
|
||||
|
||||
defer func() {
|
||||
if pipeErr == nil {
|
||||
if err := tw.Close(); err != nil {
|
||||
if !isPipeClosedError(err) {
|
||||
log.Errorf("Failed to close tar writer: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
for _, fpath := range filepaths {
|
||||
var (
|
||||
err error
|
||||
linkTarget string
|
||||
)
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
closePipeWithError(ctx.Err())
|
||||
|
||||
return
|
||||
|
||||
default:
|
||||
}
|
||||
|
||||
log.Debugf("Adding file: %s", fpath)
|
||||
|
||||
st, err := os.Lstat(fpath)
|
||||
|
||||
if err != nil {
|
||||
log.Warnf("Skipping %s: %v", fpath, err)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
if st.Mode()&os.ModeSymlink != 0 {
|
||||
linkTarget, err = os.Readlink(fpath)
|
||||
|
||||
if err != nil {
|
||||
log.Warnf("Skipping %s: readlink failed: %v", fpath, err)
|
||||
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
if st.Mode()&os.ModeSocket != 0 {
|
||||
log.Debugf("Skipping socket: %s", fpath)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
header, err := tar.FileInfoHeader(st, linkTarget)
|
||||
|
||||
if err != nil {
|
||||
log.Warnf("Skipping %s: tar header failed: %v", fpath, err)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
header.Format = tar.FormatGNU
|
||||
header.Name = strings.TrimPrefix(filepath.ToSlash(filepath.Clean(fpath)), "/")
|
||||
|
||||
mode := st.Mode()
|
||||
|
||||
if err := writeWithContext(ctx, func() error {
|
||||
return tw.WriteHeader(header)
|
||||
}); err != nil {
|
||||
closePipeWithError(err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if mode.IsRegular() {
|
||||
f, err := os.Open(fpath)
|
||||
if err != nil {
|
||||
log.Warnf("Skipping %s: open failed: %v", fpath, err)
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
err = copyWithContext(ctx, tw, f)
|
||||
|
||||
if e := f.Close(); e != nil {
|
||||
log.Warnf("Failed to close file %s: %v", fpath, e)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
log.Warnf("Skipping %s: copy failed: %v", fpath, err)
|
||||
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return pr
|
||||
}
|
||||
|
||||
// isPipeClosedError checks if the error is due to a closed pipe.
|
||||
func isPipeClosedError(err error) bool {
|
||||
if err == nil {
|
||||
|
||||
Reference in New Issue
Block a user