move DecompressStream, CompressStream to compression pkg

Signed-off-by: Sebastiaan van Stijn <[email protected]>

Author: thaJeztah

Diff for: archive.go

@@ -3,26 +3,18 @@ package archive
 
 import (
 	"archive/tar"
-	"bufio"
-	"bytes"
-	"compress/bzip2"
-	"compress/gzip"
 	"context"
 	"errors"
 	"fmt"
 	"io"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"runtime"
-	"strconv"
 	"strings"
-	"sync"
 	"syscall"
 	"time"
 
 	"github.com/containerd/log"
-	"github.com/klauspost/compress/zstd"
 	"github.com/moby/patternmatcher"
 	"github.com/moby/sys/sequential"
 	"github.com/moby/sys/user"
@@ -123,7 +115,7 @@ func IsArchivePath(path string) bool {
 		return false
 	}
 	defer file.Close()
-	rdr, err := DecompressStream(file)
+	rdr, err := compression.DecompressStream(file)
 	if err != nil {
 		return false
 	}
@@ -140,35 +132,6 @@ func DetectCompression(source []byte) compression.Compression {
 	return compression.Detect(source)
 }
 
-func xzDecompress(ctx context.Context, archive io.Reader) (io.ReadCloser, error) {
-	args := []string{"xz", "-d", "-c", "-q"}
-
-	return cmdStream(exec.CommandContext(ctx, args[0], args[1:]...), archive)
-}
-
-func gzDecompress(ctx context.Context, buf io.Reader) (io.ReadCloser, error) {
-	if noPigzEnv := os.Getenv("MOBY_DISABLE_PIGZ"); noPigzEnv != "" {
-		noPigz, err := strconv.ParseBool(noPigzEnv)
-		if err != nil {
-			log.G(ctx).WithError(err).Warn("invalid value in MOBY_DISABLE_PIGZ env var")
-		}
-		if noPigz {
-			log.G(ctx).Debugf("Use of pigz is disabled due to MOBY_DISABLE_PIGZ=%s", noPigzEnv)
-			return gzip.NewReader(buf)
-		}
-	}
-
-	unpigzPath, err := exec.LookPath("unpigz")
-	if err != nil {
-		log.G(ctx).Debugf("unpigz binary not found, falling back to go gzip library")
-		return gzip.NewReader(buf)
-	}
-
-	log.G(ctx).Debugf("Using %s to decompress", unpigzPath)
-
-	return cmdStream(exec.CommandContext(ctx, unpigzPath, "-d", "-c"), buf)
-}
-
 type readCloserWrapper struct {
 	io.Reader
 	closer func() error
@@ -181,132 +144,18 @@ func (r *readCloserWrapper) Close() error {
 	return nil
 }
 
-var bufioReader32KPool = &sync.Pool{
-	New: func() interface{} { return bufio.NewReaderSize(nil, 32*1024) },
-}
-
-type bufferedReader struct {
-	buf *bufio.Reader
-}
-
-func newBufferedReader(r io.Reader) *bufferedReader {
-	buf := bufioReader32KPool.Get().(*bufio.Reader)
-	buf.Reset(r)
-	return &bufferedReader{buf}
-}
-
-func (r *bufferedReader) Read(p []byte) (int, error) {
-	if r.buf == nil {
-		return 0, io.EOF
-	}
-	n, err := r.buf.Read(p)
-	if errors.Is(err, io.EOF) {
-		r.buf.Reset(nil)
-		bufioReader32KPool.Put(r.buf)
-		r.buf = nil
-	}
-	return n, err
-}
-
-func (r *bufferedReader) Peek(n int) ([]byte, error) {
-	if r.buf == nil {
-		return nil, io.EOF
-	}
-	return r.buf.Peek(n)
-}
-
 // DecompressStream decompresses the archive and returns a ReaderCloser with the decompressed archive.
+//
+// Deprecated: use [compression.DecompressStream].
 func DecompressStream(archive io.Reader) (io.ReadCloser, error) {
-	buf := newBufferedReader(archive)
-	bs, err := buf.Peek(10)
-	if err != nil && !errors.Is(err, io.EOF) {
-		// Note: we'll ignore any io.EOF error because there are some odd
-		// cases where the layer.tar file will be empty (zero bytes) and
-		// that results in an io.EOF from the Peek() call. So, in those
-		// cases we'll just treat it as a non-compressed stream and
-		// that means just create an empty layer.
-		// See Issue 18170
-		return nil, err
-	}
-
-	switch comp := compression.Detect(bs); comp {
-	case compression.Uncompressed:
-		return &readCloserWrapper{
-			Reader: buf,
-		}, nil
-	case compression.Gzip:
-		ctx, cancel := context.WithCancel(context.Background())
-
-		gzReader, err := gzDecompress(ctx, buf)
-		if err != nil {
-			cancel()
-			return nil, err
-		}
-		return &readCloserWrapper{
-			Reader: gzReader,
-			closer: func() error {
-				cancel()
-				return gzReader.Close()
-			},
-		}, nil
-	case compression.Bzip2:
-		bz2Reader := bzip2.NewReader(buf)
-		return &readCloserWrapper{
-			Reader: bz2Reader,
-		}, nil
-	case compression.Xz:
-		ctx, cancel := context.WithCancel(context.Background())
-
-		xzReader, err := xzDecompress(ctx, buf)
-		if err != nil {
-			cancel()
-			return nil, err
-		}
-
-		return &readCloserWrapper{
-			Reader: xzReader,
-			closer: func() error {
-				cancel()
-				return xzReader.Close()
-			},
-		}, nil
-	case compression.Zstd:
-		zstdReader, err := zstd.NewReader(buf)
-		if err != nil {
-			return nil, err
-		}
-		return &readCloserWrapper{
-			Reader: zstdReader,
-			closer: func() error {
-				zstdReader.Close()
-				return nil
-			},
-		}, nil
-	default:
-		return nil, fmt.Errorf("unsupported compression format: %s", (&comp).Extension())
-	}
+	return compression.DecompressStream(archive)
 }
 
-type nopWriteCloser struct {
-	io.Writer
-}
-
-func (nopWriteCloser) Close() error { return nil }
-
 // CompressStream compresses the dest with specified compression algorithm.
+//
+// Deprecated: use [compression.CompressStream].
 func CompressStream(dest io.Writer, comp compression.Compression) (io.WriteCloser, error) {
-	switch comp {
-	case compression.Uncompressed:
-		return nopWriteCloser{dest}, nil
-	case compression.Gzip:
-		return gzip.NewWriter(dest), nil
-	case compression.Bzip2, compression.Xz:
-		// archive/bzip2 does not support writing, and there is no xz support at all
-		// However, this is not a problem as docker only currently generates gzipped tars
-		return nil, fmt.Errorf("unsupported compression format: %s", (&comp).Extension())
-	default:
-		return nil, fmt.Errorf("unsupported compression format: %s", (&comp).Extension())
-	}
+	return compression.CompressStream(dest, comp)
 }
 
 // TarModifierFunc is a function that can be passed to ReplaceFileTarWrapper to
@@ -865,7 +714,7 @@ func NewTarballer(srcPath string, options *TarOptions) (*Tarballer, error) {
 
 	pipeReader, pipeWriter := io.Pipe()
 
-	compressWriter, err := CompressStream(pipeWriter, options.Compression)
+	compressWriter, err := compression.CompressStream(pipeWriter, options.Compression)
 	if err != nil {
 		return nil, err
 	}
@@ -1242,7 +1091,7 @@ func untarHandler(tarArchive io.Reader, dest string, options *TarOptions, decomp
 
 	r := tarArchive
 	if decompress {
-		decompressedArchive, err := DecompressStream(tarArchive)
+		decompressedArchive, err := compression.DecompressStream(tarArchive)
 		if err != nil {
 			return err
 		}
@@ -1389,43 +1238,3 @@ func remapIDs(idMapping user.IdentityMapping, hdr *tar.Header) error {
 	hdr.Uid, hdr.Gid = uid, gid
 	return err
 }
-
-// cmdStream executes a command, and returns its stdout as a stream.
-// If the command fails to run or doesn't complete successfully, an error
-// will be returned, including anything written on stderr.
-func cmdStream(cmd *exec.Cmd, input io.Reader) (io.ReadCloser, error) {
-	cmd.Stdin = input
-	pipeR, pipeW := io.Pipe()
-	cmd.Stdout = pipeW
-	var errBuf bytes.Buffer
-	cmd.Stderr = &errBuf
-
-	// Run the command and return the pipe
-	if err := cmd.Start(); err != nil {
-		return nil, err
-	}
-
-	// Ensure the command has exited before we clean anything up
-	done := make(chan struct{})
-
-	// Copy stdout to the returned pipe
-	go func() {
-		if err := cmd.Wait(); err != nil {
-			_ = pipeW.CloseWithError(fmt.Errorf("%w: %s", err, errBuf.String()))
-		} else {
-			_ = pipeW.Close()
-		}
-		close(done)
-	}()
-
-	return &readCloserWrapper{
-		Reader: pipeR,
-		closer: func() error {
-			// Close pipeR, and then wait for the command to complete before returning. We have to close pipeR first, as
-			// cmd.Wait waits for any non-file stdout/stderr/stdin to close.
-			err := pipeR.Close()
-			<-done
-			return err
-		},
-	}, nil
-}