compose/internal/sync/tar.go

/*
   Copyright 2018 The Tilt Dev Authors
   Copyright 2023 Docker Compose CLI authors

   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 sync

import (
	"archive/tar"
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"

	"github.com/moby/go-archive"
	"github.com/moby/moby/api/types/container"
	"golang.org/x/sync/errgroup"
)

type archiveEntry struct {
	path   string
	info   os.FileInfo
	header *tar.Header
}

type LowLevelClient interface {
	ContainersForService(ctx context.Context, projectName string, serviceName string) ([]container.Summary, error)

	Exec(ctx context.Context, containerID string, cmd []string, in io.Reader) error
	Untar(ctx context.Context, id string, reader io.ReadCloser) error
}

type Tar struct {
	client LowLevelClient

	projectName string
	// maxConcurrency bounds concurrent per-container sync calls. <=0
	// (including the Go zero-value) leaves it unlimited, since
	// errgroup.SetLimit(0) means "allow zero goroutines", not "unlimited".
	maxConcurrency int
}

var _ Syncer = &Tar{}

func NewTar(projectName string, client LowLevelClient, maxConcurrency int) *Tar {
	return &Tar{
		projectName:    projectName,
		client:         client,
		maxConcurrency: maxConcurrency,
	}
}

func (t *Tar) Sync(ctx context.Context, service string, paths []*PathMapping) error {
	containers, err := t.client.ContainersForService(ctx, t.projectName, service)
	if err != nil {
		return err
	}

	var pathsToCopy []PathMapping
	var pathsToDelete []string
	for _, p := range paths {
		if _, err := os.Stat(p.HostPath); err == nil {
			pathsToCopy = append(pathsToCopy, *p)
		} else if errors.Is(err, fs.ErrNotExist) {
			pathsToDelete = append(pathsToDelete, p.ContainerPath)
		} else {
			return fmt.Errorf("stat %q: %w", p.HostPath, err)
		}
	}

	var deleteCmd []string
	if len(pathsToDelete) != 0 {
		deleteCmd = append([]string{"rm", "-rf"}, pathsToDelete...)
	}

	var (
		eg    errgroup.Group
		errMu sync.Mutex
		errs  = make([]error, 0, len(containers)*2) // max 2 errs per container
	)

	if t.maxConcurrency > 0 {
		eg.SetLimit(t.maxConcurrency)
	}
	for i := range containers {
		containerID := containers[i].ID
		tarReader := tarArchive(pathsToCopy, keepImpliedDirectories)

		eg.Go(func() error {
			if len(deleteCmd) != 0 {
				if err := t.client.Exec(ctx, containerID, deleteCmd, nil); err != nil {
					errMu.Lock()
					errs = append(errs, fmt.Errorf("deleting paths in %s: %w", containerID, err))
					errMu.Unlock()
				}
			}

			if err := t.client.Untar(ctx, containerID, tarReader); err != nil {
				// The engine answers with a plain 500, so the message is the only discriminator.
				if !strings.Contains(err.Error(), "cannot overwrite non-directory") {
					errMu.Lock()
					errs = append(errs, fmt.Errorf("copying files to %s: %w", containerID, err))
					errMu.Unlock()
					return nil
				}

				// A directory header for a path the container resolves through a symlink makes
				// the engine reject the whole copy, so retry without the headers the archive's
				// own entries already imply. That archive is a strict subset of the one that
				// just failed, so sending it again is safe.
				retry := tarArchive(pathsToCopy, dropImpliedDirectories)
				if retryErr := t.client.Untar(ctx, containerID, retry); retryErr != nil {
					errMu.Lock()
					errs = append(errs, fmt.Errorf("copying files to %s: %w", containerID, errors.Join(err, retryErr)))
					errMu.Unlock()
				}
			}
			return nil // don't fail-fast; collect all errors
		})
	}

	_ = eg.Wait()
	return errors.Join(errs...)
}

type ArchiveBuilder struct {
	tw *tar.Writer
	// A shared I/O buffer to help with file copying.
	copyBuf *bytes.Buffer
}

func NewArchiveBuilder(writer io.Writer) *ArchiveBuilder {
	tw := tar.NewWriter(writer)
	return &ArchiveBuilder{
		tw:      tw,
		copyBuf: &bytes.Buffer{},
	}
}

func (a *ArchiveBuilder) Close() error {
	return a.tw.Close()
}

// ArchivePathsIfExist creates a tar archive of all local files in `paths`. It quietly skips any paths that don't exist.
func (a *ArchiveBuilder) ArchivePathsIfExist(paths []PathMapping, dropImpliedDirs bool) error {
	// In order to handle overlapping syncs, we
	// 1) collect all the entries,
	// 2) de-dupe them, with last-one-wins semantics
	// 3) write all the entries
	//
	// It's not obvious that this is the correct behavior. A better approach
	// (that's more in-line with how syncs work) might ignore files in earlier
	// path mappings when we know they're going to be "synced" over.
	// There's a bunch of subtle product decisions about how overlapping path
	// mappings work that we're not sure about.
	var entries []archiveEntry
	for _, p := range paths {
		newEntries, err := a.entriesForPath(p.HostPath, p.ContainerPath)
		if err != nil {
			return fmt.Errorf("inspecting %q: %w", p.HostPath, err)
		}

		entries = append(entries, newEntries...)
	}

	entries = dedupeEntries(entries)
	if dropImpliedDirs {
		entries = withoutImpliedDirectories(entries)
	}
	for _, entry := range entries {
		err := a.writeEntry(entry)
		if err != nil {
			return fmt.Errorf("archiving %q: %w", entry.path, err)
		}
	}
	return nil
}

func (a *ArchiveBuilder) writeEntry(entry archiveEntry) error {
	pathInTar := entry.path
	header := entry.header

	if header.Typeflag != tar.TypeReg {
		// anything other than a regular file (e.g. dir, symlink) just needs the header
		if err := a.tw.WriteHeader(header); err != nil {
			return fmt.Errorf("writing %q header: %w", pathInTar, err)
		}
		return nil
	}

	file, err := os.Open(pathInTar)
	if err != nil {
		// In case the file has been deleted since we last looked at it.
		if os.IsNotExist(err) {
			return nil
		}
		return err
	}

	defer func() {
		_ = file.Close()
	}()

	// The size header must match the number of contents bytes.
	//
	// There is room for a race condition here if something writes to the file
	// after we've read the file size.
	//
	// For small files, we avoid this by first copying the file into a buffer,
	// and using the size of the buffer to populate the header.
	//
	// For larger files, we don't want to copy the whole thing into a buffer,
	// because that would blow up heap size. There is some danger that this
	// will lead to a spurious error when the tar writer validates the sizes.
	// That error will be disruptive but will be handled as best as we
	// can downstream.
	useBuf := header.Size < 5000000
	if useBuf {
		a.copyBuf.Reset()
		_, err = io.Copy(a.copyBuf, file)
		if err != nil && !errors.Is(err, io.EOF) {
			return fmt.Errorf("copying %q: %w", pathInTar, err)
		}
		header.Size = int64(len(a.copyBuf.Bytes()))
	}

	// wait to write the header until _after_ the file is successfully opened
	// to avoid generating an invalid tar entry that has a header but no contents
	// in the case the file has been deleted
	err = a.tw.WriteHeader(header)
	if err != nil {
		return fmt.Errorf("writing %q header: %w", pathInTar, err)
	}

	if useBuf {
		_, err = io.Copy(a.tw, a.copyBuf)
	} else {
		_, err = io.Copy(a.tw, file)
	}

	if err != nil && !errors.Is(err, io.EOF) {
		return fmt.Errorf("copying %q: %w", pathInTar, err)
	}

	// explicitly flush so that if the entry is invalid we will detect it now and
	// provide a more meaningful error
	if err := a.tw.Flush(); err != nil {
		return fmt.Errorf("finalizing %q: %w", pathInTar, err)
	}
	return nil
}

// entriesForPath writes the given source path into tarWriter at the given dest (recursively for directories).
// e.g. tarring my_dir --> dest d: d/file_a, d/file_b
// If source path does not exist, quietly skips it and returns no err
func (a *ArchiveBuilder) entriesForPath(localPath, containerPath string) ([]archiveEntry, error) {
	localInfo, err := os.Stat(localPath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, err
	}

	localPathIsDir := localInfo.IsDir()
	if localPathIsDir {
		// Make sure we can trim this off filenames to get valid relative filepaths
		if !strings.HasSuffix(localPath, string(filepath.Separator)) {
			localPath += string(filepath.Separator)
		}
	}

	containerPath = strings.TrimPrefix(containerPath, "/")

	result := make([]archiveEntry, 0)
	err = filepath.Walk(localPath, func(curLocalPath string, info os.FileInfo, err error) error {
		if err != nil {
			return fmt.Errorf("walking %q: %w", curLocalPath, err)
		}

		linkname := ""
		if info.Mode()&os.ModeSymlink != 0 {
			var err error
			linkname, err = os.Readlink(curLocalPath)
			if err != nil {
				return err
			}
		}

		var name string
		//nolint:gocritic
		if localPathIsDir {
			// Name of file in tar should be relative to source directory...
			tmp, err := filepath.Rel(localPath, curLocalPath)
			if err != nil {
				return fmt.Errorf("making %q relative to %q: %w", curLocalPath, localPath, err)
			}
			// ...and live inside `dest`
			name = path.Join(containerPath, filepath.ToSlash(tmp))
		} else if strings.HasSuffix(containerPath, "/") {
			name = containerPath + filepath.Base(curLocalPath)
		} else {
			name = containerPath
		}

		header, err := archive.FileInfoHeader(name, info, linkname)
		if err != nil {
			// Not all types of files are allowed in a tarball. That's OK.
			// Mimic the Docker behavior and just skip the file.
			return nil
		}

		result = append(result, archiveEntry{
			path:   curLocalPath,
			info:   info,
			header: header,
		})

		return nil
	})
	if err != nil {
		return nil, err
	}
	return result, nil
}

// Whether tarArchive keeps the header of a directory the archive itself fills with entries.
const (
	keepImpliedDirectories = false
	dropImpliedDirectories = true
)

func tarArchive(ops []PathMapping, dropImpliedDirs bool) io.ReadCloser {
	pr, pw := io.Pipe()
	go func() {
		ab := NewArchiveBuilder(pw)
		err := ab.ArchivePathsIfExist(ops, dropImpliedDirs)
		if err != nil {
			_ = pw.CloseWithError(fmt.Errorf("adding files to tar: %w", err))
		} else {
			// propagate errors from the TarWriter::Close() because it performs a final
			// Flush() and any errors mean the tar is invalid
			if err := ab.Close(); err != nil {
				_ = pw.CloseWithError(fmt.Errorf("closing tar: %w", err))
			} else {
				_ = pw.Close()
			}
		}
	}()
	return pr
}

// withoutImpliedDirectories removes the header of every directory the archive already fills with
// entries: the engine creates those while extracting, while an explicit header makes it reject the
// whole copy when the container resolves that path through a symlink. A directory nothing else
// would create - an empty one - keeps its header.
//
// The engine creates them with ImpliedDirectoryMode and its own root uid/gid rather than the host's,
// so this is a fallback for a copy that already failed, never the first attempt.
func withoutImpliedDirectories(entries []archiveEntry) []archiveEntry {
	holdsEntries := make(map[string]bool, len(entries))
	for _, entry := range entries {
		holdsEntries[path.Dir(path.Clean(entry.header.Name))] = true
	}
	result := make([]archiveEntry, 0, len(entries))
	for _, entry := range entries {
		if entry.header.Typeflag == tar.TypeDir && holdsEntries[path.Clean(entry.header.Name)] {
			continue
		}
		result = append(result, entry)
	}
	return result
}

// Dedupe the entries with last-entry-wins semantics.
func dedupeEntries(entries []archiveEntry) []archiveEntry {
	seenIndex := make(map[string]int, len(entries))
	result := make([]archiveEntry, 0, len(entries))
	for i, entry := range entries {
		seenIndex[entry.header.Name] = i
	}
	for i, entry := range entries {
		if seenIndex[entry.header.Name] == i {
			result = append(result, entry)
		}
	}
	return result
}