Wrap public transport implementations to consistently use a private API

copy/copy.go

@@ -15,6 +15,8 @@ import (
 	"github.com/containers/image/v5/docker/reference"
 	"github.com/containers/image/v5/image"
 	internalblobinfocache "github.com/containers/image/v5/internal/blobinfocache"
+	"github.com/containers/image/v5/internal/imagedestination"
+	"github.com/containers/image/v5/internal/imagesource"
 	"github.com/containers/image/v5/internal/pkg/platform"
 	"github.com/containers/image/v5/internal/private"
 	"github.com/containers/image/v5/manifest"
@@ -63,8 +65,8 @@ var expectedCompressionFormats = map[string]*compressiontypes.Algorithm{
 // copier allows us to keep track of diffID values for blobs, and other
 // data shared across one or more images in a possible manifest list.
 type copier struct {
-	dest                          types.ImageDestination
-	rawSource                     types.ImageSource
+	dest                          private.ImageDestination
+	rawSource                     private.ImageSource
 	reportWriter                  io.Writer
 	progressOutput                io.Writer
 	progressInterval              time.Duration
@@ -202,20 +204,22 @@ func Image(ctx context.Context, policyContext *signature.PolicyContext, destRef,
 		reportWriter = options.ReportWriter
 	}
 
-	dest, err := destRef.NewImageDestination(ctx, options.DestinationCtx)
+	publicDest, err := destRef.NewImageDestination(ctx, options.DestinationCtx)
 	if err != nil {
 		return nil, errors.Wrapf(err, "initializing destination %s", transports.ImageName(destRef))
 	}
+	dest := imagedestination.FromPublic(publicDest)
 	defer func() {
 		if err := dest.Close(); err != nil {
 			retErr = errors.Wrapf(retErr, " (dest: %v)", err)
 		}
 	}()
 
-	rawSource, err := srcRef.NewImageSource(ctx, options.SourceCtx)
+	publicRawSource, err := srcRef.NewImageSource(ctx, options.SourceCtx)
 	if err != nil {
 		return nil, errors.Wrapf(err, "initializing source %s", transports.ImageName(srcRef))
 	}
+	rawSource := imagesource.FromPublic(publicRawSource)
 	defer func() {
 		if err := rawSource.Close(); err != nil {
 			retErr = errors.Wrapf(retErr, " (src: %v)", err)
@@ -1225,28 +1229,13 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 		// a failure when we eventually try to update the manifest with the digest and MIME type of the reused blob.
 		// Fixing that will probably require passing more information to TryReusingBlob() than the current version of
 		// the ImageDestination interface lets us pass in.
-		var (
-			blobInfo types.BlobInfo
-			reused   bool
-			err      error
-		)
-		// Note: the storage destination optimizes the committing of
-		// layers which requires passing the index of the layer.
-		// Hence, we need to special case and cast.
-		dest, ok := ic.c.dest.(private.ImageDestination)
-		if ok {
-			options := private.TryReusingBlobOptions{
-				Cache:         ic.c.blobInfoCache,
-				CanSubstitute: ic.canSubstituteBlobs,
-				SrcRef:        srcRef,
-				EmptyLayer:    emptyLayer,
-			}
-			options.LayerIndex = &layerIndex
-			reused, blobInfo, err = dest.TryReusingBlobWithOptions(ctx, srcInfo, options)
-		} else {
-			reused, blobInfo, err = ic.c.dest.TryReusingBlob(ctx, srcInfo, ic.c.blobInfoCache, ic.canSubstituteBlobs)
-		}
-
+		reused, blobInfo, err := ic.c.dest.TryReusingBlobWithOptions(ctx, srcInfo, private.TryReusingBlobOptions{
+			Cache:         ic.c.blobInfoCache,
+			CanSubstitute: ic.canSubstituteBlobs,
+			EmptyLayer:    emptyLayer,
+			LayerIndex:    &layerIndex,
+			SrcRef:        srcRef,
+		})
 		if err != nil {
 			return types.BlobInfo{}, "", errors.Wrapf(err, "trying to reuse blob %s at destination", srcInfo.Digest)
 		}
@@ -1286,9 +1275,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 	// of the source file are not known yet and must be fetched.
 	// Attempt a partial only when the source allows to retrieve a blob partially and
 	// the destination has support for it.
-	imgSource, okSource := ic.c.rawSource.(private.ImageSourceSeekable)
-	imgDest, okDest := ic.c.dest.(private.ImageDestinationPartial)
-	if okSource && okDest && !diffIDIsNeeded {
+	if ic.c.rawSource.SupportsGetBlobAt() && ic.c.dest.SupportsPutBlobPartial() && !diffIDIsNeeded {
 		if reused, blobInfo := func() (bool, types.BlobInfo) { // A scope for defer
 			bar := ic.c.createProgressBar(pool, true, srcInfo, "blob", "done")
 			hideProgressBar := true
@@ -1302,8 +1289,8 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 			defer close(terminate)
 			defer close(progress)
 
-			proxy := imageSourceSeekableProxy{
-				source:   imgSource,
+			proxy := blobChunkAccessorProxy{
+				wrapped:  ic.c.rawSource,
 				progress: progress,
 			}
 			go func() {
@@ -1318,7 +1305,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 			}()
 
 			bar.SetTotal(srcInfo.Size, false)
-			info, err := imgDest.PutBlobPartial(ctx, proxy, srcInfo, ic.c.blobInfoCache)
+			info, err := ic.c.dest.PutBlobPartial(ctx, proxy, srcInfo, ic.c.blobInfoCache)
 			if err == nil {
 				bar.SetRefill(srcInfo.Size - bar.Current())
 				bar.SetCurrent(srcInfo.Size)
@@ -1658,24 +1645,15 @@ func (c *copier) copyBlobFromStream(ctx context.Context, srcStream io.Reader, sr
 	}
 
 	// === Finally, send the layer stream to dest.
-	var uploadedInfo types.BlobInfo
-	// Note: the storage destination optimizes the committing of layers
-	// which requires passing the index of the layer.  Hence, we need to
-	// special case and cast.
-	dest, ok := c.dest.(private.ImageDestination)
-	if ok {
-		options := private.PutBlobOptions{
-			Cache:      c.blobInfoCache,
-			IsConfig:   isConfig,
-			EmptyLayer: emptyLayer,
-		}
-		if !isConfig {
-			options.LayerIndex = &layerIndex
-		}
-		uploadedInfo, err = dest.PutBlobWithOptions(ctx, &errorAnnotationReader{destStream}, inputInfo, options)
-	} else {
-		uploadedInfo, err = c.dest.PutBlob(ctx, &errorAnnotationReader{destStream}, inputInfo, c.blobInfoCache, isConfig)
+	options := private.PutBlobOptions{
+		Cache:      c.blobInfoCache,
+		IsConfig:   isConfig,
+		EmptyLayer: emptyLayer,
+	}
+	if !isConfig {
+		options.LayerIndex = &layerIndex
 	}
+	uploadedInfo, err := c.dest.PutBlobWithOptions(ctx, &errorAnnotationReader{destStream}, inputInfo, options)
 	if err != nil {
 		return types.BlobInfo{}, errors.Wrap(err, "writing blob")
 	}

copy/progress_reader.go

@@ -80,19 +80,22 @@ func (r *progressReader) Read(p []byte) (int, error) {
 	return n, err
 }
 
-// imageSourceSeekableProxy wraps ImageSourceSeekable and keeps track of how many bytes
+// blobChunkAccessorProxy wraps a BlobChunkAccessor and keeps track of how many bytes
 // are received.
-type imageSourceSeekableProxy struct {
-	// source is the seekable input to read from.
-	source private.ImageSourceSeekable
+type blobChunkAccessorProxy struct {
+	// wrapped is the underlying BlobChunkAccessor
+	wrapped private.BlobChunkAccessor
 	// progress is the chan where the total number of bytes read so far are reported.
 	progress chan int64
 }
 
-// GetBlobAt reads from the ImageSourceSeekable and report how many bytes were received
-// to the progress chan.
-func (s imageSourceSeekableProxy) GetBlobAt(ctx context.Context, bInfo types.BlobInfo, chunks []private.ImageSourceChunk) (chan io.ReadCloser, chan error, error) {
-	rc, errs, err := s.source.GetBlobAt(ctx, bInfo, chunks)
+// GetBlobAt returns a sequential channel of readers that contain data for the requested
+// blob chunks, and a channel that might get a single error value.
+// The specified chunks must be not overlapping and sorted by their offset.
+// The readers must be fully consumed, in the order they are returned, before blocking
+// to read the next chunk.
+func (s blobChunkAccessorProxy) GetBlobAt(ctx context.Context, info types.BlobInfo, chunks []private.ImageSourceChunk) (chan io.ReadCloser, chan error, error) {
+	rc, errs, err := s.wrapped.GetBlobAt(ctx, info, chunks)
 	if err == nil {
 		total := int64(0)
 		for _, c := range chunks {

copy/sign_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/containers/image/v5/directory"
 	"github.com/containers/image/v5/docker"
+	"github.com/containers/image/v5/internal/imagedestination"
 	"github.com/containers/image/v5/manifest"
 	"github.com/containers/image/v5/signature"
 	"github.com/containers/image/v5/types"
@@ -47,7 +48,7 @@ func TestCreateSignature(t *testing.T) {
 	require.NoError(t, err)
 	defer dirDest.Close()
 	c := &copier{
-		dest:         dirDest,
+		dest:         imagedestination.FromPublic(dirDest),
 		reportWriter: ioutil.Discard,
 	}
 	_, err = c.createSignature(manifestBlob, testKeyFingerprint, "")
@@ -61,7 +62,7 @@ func TestCreateSignature(t *testing.T) {
 	require.NoError(t, err)
 	defer dockerDest.Close()
 	c = &copier{
-		dest:         dockerDest,
+		dest:         imagedestination.FromPublic(dockerDest),
 		reportWriter: ioutil.Discard,
 	}
 

docker/docker_image_src.go

@@ -147,6 +147,11 @@ func (s *dockerImageSource) Close() error {
 	return nil
 }
 
+// SupportsGetBlobAt() returns true if GetBlobAt (BlobChunkAccessor) is supported.
+func (s *dockerImageSource) SupportsGetBlobAt() bool {
+	return true
+}
+
 // LayerInfosForCopy returns either nil (meaning the values in the manifest are fine), or updated values for the layer
 // blobsums that are listed in the image's manifest.  If values are returned, they should be used when using GetBlob()
 // to read the image's layers.
@@ -357,8 +362,11 @@ func handle206Response(streams chan io.ReadCloser, errs chan error, body io.Read
 	}
 }
 
-// GetBlobAt returns a stream for the specified blob.
+// GetBlobAt returns a sequential channel of readers that contain data for the requested
+// blob chunks, and a channel that might get a single error value.
 // The specified chunks must be not overlapping and sorted by their offset.
+// The readers must be fully consumed, in the order they are returned, before blocking
+// to read the next chunk.
 func (s *dockerImageSource) GetBlobAt(ctx context.Context, info types.BlobInfo, chunks []private.ImageSourceChunk) (chan io.ReadCloser, chan error, error) {
 	headers := make(map[string][]string)
 

docker/docker_image_src_test.go

@@ -19,7 +19,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-var _ private.ImageSourceSeekable = (*dockerImageSource)(nil)
+var _ private.ImageSource = (*dockerImageSource)(nil)
 
 func TestDockerImageSourceReference(t *testing.T) {
 	manifestPathRegex := regexp.MustCompile("^/v2/.*/manifests/latest$")

internal/imagedestination/wrapper.go

@@ -0,0 +1,69 @@
+package imagedestination
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/containers/image/v5/internal/private"
+	"github.com/containers/image/v5/types"
+)
+
+// FromPublic(dest) returns an object that provides the private.ImageDestination API
+//
+// Eventually, we might want to expose this function, and methods of the returned object,
+// as a public API (or rather, a variant that does not include the already-superseded
+// methods of types.ImageDestination, and has added more future-proofing), and more strongly
+// deprecate direct use of types.ImageDestination.
+//
+// NOTE: The returned API MUST NOT be a public interface (it can be either just a struct
+// with public methods, or perhaps a private interface), so that we can add methods
+// without breaking any external implementors of a public interface.
+func FromPublic(dest types.ImageDestination) private.ImageDestination {
+	if dest2, ok := dest.(private.ImageDestination); ok {
+		return dest2
+	}
+	return &wrapped{ImageDestination: dest}
+}
+
+// wrapped provides the private.ImageDestination operations
+// for a destination that only implements types.ImageDestination
+type wrapped struct {
+	types.ImageDestination
+}
+
+// SupportsPutBlobPartial returns true if PutBlobPartial is supported.
+func (w *wrapped) SupportsPutBlobPartial() bool {
+	return false
+}
+
+// PutBlobWithOptions writes contents of stream and returns data representing the result.
+// inputInfo.Digest can be optionally provided if known; if provided, and stream is read to the end without error, the digest MUST match the stream contents.
+// inputInfo.Size is the expected length of stream, if known.
+// inputInfo.MediaType describes the blob format, if known.
+// WARNING: The contents of stream are being verified on the fly.  Until stream.Read() returns io.EOF, the contents of the data SHOULD NOT be available
+// to any other readers for download using the supplied digest.
+// If stream.Read() at any time, ESPECIALLY at end of input, returns an error, PutBlob MUST 1) fail, and 2) delete any data stored so far.
+func (w *wrapped) PutBlobWithOptions(ctx context.Context, stream io.Reader, inputInfo types.BlobInfo, options private.PutBlobOptions) (types.BlobInfo, error) {
+	return w.PutBlob(ctx, stream, inputInfo, options.Cache, options.IsConfig)
+}
+
+// PutBlobPartial attempts to create a blob using the data that is already present
+// at the destination. chunkAccessor is accessed in a non-sequential way to retrieve the missing chunks.
+// It is available only if SupportsPutBlobPartial().
+// Even if SupportsPutBlobPartial() returns true, the call can fail, in which case the caller
+// should fall back to PutBlobWithOptions.
+func (w *wrapped) PutBlobPartial(ctx context.Context, chunkAccessor private.BlobChunkAccessor, srcInfo types.BlobInfo, cache types.BlobInfoCache) (types.BlobInfo, error) {
+	return types.BlobInfo{}, fmt.Errorf("internal error: PutBlobPartial is not supported by the %q transport", w.Reference().Transport().Name())
+}
+
+// TryReusingBlobWithOptions checks whether the transport already contains, or can efficiently reuse, a blob, and if so, applies it to the current destination
+// (e.g. if the blob is a filesystem layer, this signifies that the changes it describes need to be applied again when composing a filesystem tree).
+// info.Digest must not be empty.
+// If the blob has been successfully reused, returns (true, info, nil); info must contain at least a digest and size, and may
+// include CompressionOperation and CompressionAlgorithm fields to indicate that a change to the compression type should be
+// reflected in the manifest that will be written.
+// If the transport can not reuse the requested blob, TryReusingBlob returns (false, {}, nil); it returns a non-nil error only on an unexpected failure.
+func (w *wrapped) TryReusingBlobWithOptions(ctx context.Context, info types.BlobInfo, options private.TryReusingBlobOptions) (bool, types.BlobInfo, error) {
+	return w.TryReusingBlob(ctx, info, options.Cache, options.CanSubstitute)
+}

internal/imagesource/wrapper.go

@@ -0,0 +1,47 @@
+package imagesource
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/containers/image/v5/internal/private"
+	"github.com/containers/image/v5/types"
+)
+
+// FromPublic(src) returns an object that provides the private.ImageSource API
+//
+// Eventually, we might want to expose this function, and methods of the returned object,
+// as a public API (or rather, a variant that does not include the already-superseded
+// methods of types.ImageSource, and has added more future-proofing), and more strongly
+// deprecate direct use of types.ImageSource.
+//
+// NOTE: The returned API MUST NOT be a public interface (it can be either just a struct
+// with public methods, or perhaps a private interface), so that we can add methods
+// without breaking any external implementors of a public interface.
+func FromPublic(src types.ImageSource) private.ImageSource {
+	if src2, ok := src.(private.ImageSource); ok {
+		return src2
+	}
+	return &wrapped{ImageSource: src}
+}
+
+// wrapped provides the private.ImageSource operations
+// for a source that only implements types.ImageSource
+type wrapped struct {
+	types.ImageSource
+}
+
+// SupportsGetBlobAt() returns true if GetBlobAt (BlobChunkAccessor) is supported.
+func (w *wrapped) SupportsGetBlobAt() bool {
+	return false
+}
+
+// GetBlobAt returns a sequential channel of readers that contain data for the requested
+// blob chunks, and a channel that might get a single error value.
+// The specified chunks must be not overlapping and sorted by their offset.
+// The readers must be fully consumed, in the order they are returned, before blocking
+// to read the next chunk.
+func (w *wrapped) GetBlobAt(ctx context.Context, info types.BlobInfo, chunks []private.ImageSourceChunk) (chan io.ReadCloser, chan error, error) {
+	return nil, nil, fmt.Errorf("internal error: GetBlobAt is not supported by the %q transport", w.Reference().Transport().Name())
+}