Use Temp File For layout.{Write,Append,Replace} Image/Index Methods (…

…#1226)

* Add layout.WriteLayer for streaming layers and undersized layers from incomplete downloads

WriteLayer is an alternative to WriteBlob which can be used with
streaming layers which have not yet calculated a digest or size. A
temporary file is first written and then renamed when the reader has
been fully consumed. Also, existing layers which do not have the correct
size (and thus invalid hashes, but the hash is not checked) are
overwritten.

{Append,Replace,Write}{Image,Layer} functions utilize WriteLayer instead
of WriteBlob to correctly overwrite undersized layers which were
previously ignored without error. Undersized layers should no longer
occur, since WriteLayer writes to a temporary file first, but when
WriteBlob was used, these were common due to incomplete downloads.

* Improve error messages for validate.Image and validate.Index when layer blob content is truncated

* Unexport `layout.WriteLayer`; Implement suggested changes for validateLayers and temp file naming

* Add tests for closing stream.Layer

* Check size of blob written when expected size is known

* Clean up temp file when layout.WriteBlob errors

* Refactor `stream.compressedReader` to have less access to mutable fields and fix data race

* Panic when `(layout.Path).writeBlob` is called with invalid arguments

* Warn when temp file cleanup fails after failing to write a blob

pkg/v1/layout/write.go

@@ -17,15 +17,19 @@ package layout
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 
+	"github.com/google/go-containerregistry/pkg/logs"
 	v1 "github.com/google/go-containerregistry/pkg/v1"
 	"github.com/google/go-containerregistry/pkg/v1/match"
 	"github.com/google/go-containerregistry/pkg/v1/mutate"
 	"github.com/google/go-containerregistry/pkg/v1/partial"
+	"github.com/google/go-containerregistry/pkg/v1/stream"
 	"github.com/google/go-containerregistry/pkg/v1/types"
 	"golang.org/x/sync/errgroup"
 )
@@ -221,34 +225,95 @@ func (l Path) WriteFile(name string, data []byte, perm os.FileMode) error {
 // WriteBlob copies a file to the blobs/ directory in the Path from the given ReadCloser at
 // blobs/{hash.Algorithm}/{hash.Hex}.
 func (l Path) WriteBlob(hash v1.Hash, r io.ReadCloser) error {
+	return l.writeBlob(hash, -1, r, nil)
+}
+
+func (l Path) writeBlob(hash v1.Hash, size int64, r io.Reader, renamer func() (v1.Hash, error)) error {
+	if hash.Hex == "" && renamer == nil {
+		panic("writeBlob called an invalid hash and no renamer")
+	}
+
 	dir := l.path("blobs", hash.Algorithm)
 	if err := os.MkdirAll(dir, os.ModePerm); err != nil && !os.IsExist(err) {
 		return err
 	}
 
+	// Check if blob already exists and is the correct size
 	file := filepath.Join(dir, hash.Hex)
-	if _, err := os.Stat(file); err == nil {
-		// Blob already exists, that's fine.
+	if s, err := os.Stat(file); err == nil && !s.IsDir() && (s.Size() == size || size == -1) {
 		return nil
 	}
-	w, err := os.Create(file)
+
+	// If a renamer func was provided write to a temporary file
+	open := func() (*os.File, error) { return os.Create(file) }
+	if renamer != nil {
+		open = func() (*os.File, error) { return ioutil.TempFile(dir, hash.Hex) }
+	}
+	w, err := open()
 	if err != nil {
 		return err
 	}
+	if renamer != nil {
+		// Delete temp file if an error is encountered before renaming
+		defer func() {
+			if err := os.Remove(w.Name()); err != nil && !errors.Is(err, os.ErrNotExist) {
+				logs.Warn.Printf("error removing temporary file after encountering an error while writing blob: %v", err)
+			}
+		}()
+	}
 	defer w.Close()
 
-	_, err = io.Copy(w, r)
-	return err
-}
+	// Write to file and exit if not renaming
+	if n, err := io.Copy(w, r); err != nil || renamer == nil {
+		return err
+	} else if size != -1 && n != size {
+		return fmt.Errorf("expected blob size %d, but only wrote %d", size, n)
+	}
+
+	// Always close file before renaming
+	if err := w.Close(); err != nil {
+		return err
+	}
+
+	// Rename file based on the final hash
+	finalHash, err := renamer()
+	if err != nil {
+		return fmt.Errorf("error getting final digest of layer: %w", err)
+	}
 
-// TODO: A streaming version of WriteBlob so we don't have to know the hash
-// before we write it.
+	renamePath := l.path("blobs", finalHash.Algorithm, finalHash.Hex)
+	return os.Rename(w.Name(), renamePath)
+}
 
-// TODO: For streaming layers we should write to a tmp file then Rename to the
-// final digest.
+// writeLayer writes the compressed layer to a blob. Unlike WriteBlob it will
+// write to a temporary file (suffixed with .tmp) within the layout until the
+// compressed reader is fully consumed and written to disk. Also unlike
+// WriteBlob, it will not skip writing and exit without error when a blob file
+// exists, but does not have the correct size. (The blob hash is not
+// considered, because it may be expensive to compute.)
 func (l Path) writeLayer(layer v1.Layer) error {
 	d, err := layer.Digest()
-	if err != nil {
+	if errors.Is(err, stream.ErrNotComputed) {
+		// Allow digest errors, since streams may not have calculated the hash
+		// yet. Instead, use an empty value, which will be transformed into a
+		// random file name with `ioutil.TempFile` and the final digest will be
+		// calculated after writing to a temp file and before renaming to the
+		// final path.
+		d = v1.Hash{Algorithm: "sha256", Hex: ""}
+	} else if err != nil {
+		return err
+	}
+
+	s, err := layer.Size()
+	if errors.Is(err, stream.ErrNotComputed) {
+		// Allow size errors, since streams may not have calculated the size
+		// yet. Instead, use zero as a sentinel value meaning that no size
+		// comparison can be done and any sized blob file should be considered
+		// valid and not overwritten.
+		//
+		// TODO: Provide an option to always overwrite blobs.
+		s = -1
+	} else if err != nil {
 		return err
 	}
 
@@ -257,7 +322,10 @@ func (l Path) writeLayer(layer v1.Layer) error {
 		return err
 	}
 
-	return l.WriteBlob(d, r)
+	if err := l.writeBlob(d, s, r, layer.Digest); err != nil {
+		return fmt.Errorf("error writing layer: %w", err)
+	}
+	return nil
 }
 
 // RemoveBlob removes a file from the blobs directory in the Path