Add delta layer support for efficient image pulls (authored by Claude)

Signed-off-by: Vance Raiti <[email protected]>

Author: vraiti

image/copy/compression.go

@@ -120,6 +120,7 @@ func (ic *imageCopier) blobPipelineCompressionStep(stream *sourceStream, canModi
 	if canModifyBlob && layerCompressionChangeSupported {
 		for _, fn := range []func(*sourceStream, bpDetectCompressionStepData) (*bpCompressionStepData, error){
 			ic.bpcPreserveEncrypted,
+			ic.bpcPreserveNoCompress,
 			ic.bpcCompressUncompressed,
 			ic.bpcRecompressCompressed,
 			ic.bpcDecompressCompressed,
@@ -153,6 +154,22 @@ func (ic *imageCopier) bpcPreserveEncrypted(stream *sourceStream, _ bpDetectComp
 	return nil, nil
 }
 
+// bpcPreserveNoCompress checks if the input is a no-compress type (like tar-diff), and returns a *bpCompressionStepData if so.
+func (ic *imageCopier) bpcPreserveNoCompress(stream *sourceStream, _ bpDetectCompressionStepData) (*bpCompressionStepData, error) {
+	if manifest.IsNoCompressType(stream.info.MediaType) {
+		logrus.Debugf("Using original blob without modification for no-compress type")
+		return &bpCompressionStepData{
+			operation:                             bpcOpPreserveOpaque,
+			uploadedOperation:                     types.PreserveOriginal,
+			uploadedAlgorithm:                     nil,
+			srcCompressorBaseVariantName:          internalblobinfocache.UnknownCompression,
+			uploadedCompressorBaseVariantName:     internalblobinfocache.UnknownCompression,
+			uploadedCompressorSpecificVariantName: internalblobinfocache.UnknownCompression,
+		}, nil
+	}
+	return nil, nil
+}
+
 // bpcCompressUncompressed checks if we should be compressing an uncompressed input, and returns a *bpCompressionStepData if so.
 func (ic *imageCopier) bpcCompressUncompressed(stream *sourceStream, detected bpDetectCompressionStepData) (*bpCompressionStepData, error) {
 	if ic.c.dest.DesiredLayerCompression() == types.Compress && !detected.isCompressed {

image/copy/single.go

@@ -10,9 +10,11 @@ import (
 	"maps"
 	"reflect"
 	"slices"
+	"sort"
 	"strings"
 	"sync"
 
+	tarpatch "github.com/containers/tar-diff/pkg/tar-patch"
 	digest "github.com/opencontainers/go-digest"
 	imgspecv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/sirupsen/logrus"
@@ -448,6 +450,11 @@ func (ic *imageCopier) copyLayers(ctx context.Context) ([]compressiontypes.Algor
 		srcInfosUpdated = true
 	}
 
+	deltaLayers, err := types.ImageDeltaLayers(ic.src, ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	type copyLayerData struct {
 		destInfo types.BlobInfo
 		diffID   digest.Digest
@@ -466,7 +473,7 @@ func (ic *imageCopier) copyLayers(ctx context.Context) ([]compressiontypes.Algor
 	copyGroup := sync.WaitGroup{}
 
 	data := make([]copyLayerData, len(srcInfos))
-	copyLayerHelper := func(index int, srcLayer types.BlobInfo, toEncrypt bool, pool *mpb.Progress, srcRef reference.Named) {
+	copyLayerHelper := func(index int, srcLayer types.BlobInfo, toEncrypt bool, pool *mpb.Progress, srcRef reference.Named, deltaLayers []types.BlobInfo) {
 		defer ic.c.concurrentBlobCopiesSemaphore.Release(1)
 		defer copyGroup.Done()
 		cld := copyLayerData{}
@@ -481,7 +488,7 @@ func (ic *imageCopier) copyLayers(ctx context.Context) ([]compressiontypes.Algor
 				logrus.Debugf("Skipping foreign layer %q copy to %s", cld.destInfo.Digest, ic.c.dest.Reference().Transport().Name())
 			}
 		} else {
-			cld.destInfo, cld.diffID, cld.err = ic.copyLayer(ctx, srcLayer, toEncrypt, pool, index, srcRef, manifestLayerInfos[index].EmptyLayer)
+			cld.destInfo, cld.diffID, cld.err = ic.copyLayer(ctx, index, srcLayer, toEncrypt, pool, srcRef, manifestLayerInfos[index].EmptyLayer, deltaLayers)
 		}
 		data[index] = cld
 	}
@@ -521,7 +528,7 @@ func (ic *imageCopier) copyLayers(ctx context.Context) ([]compressiontypes.Algor
 				return fmt.Errorf("copying layer: %w", err)
 			}
 			copyGroup.Add(1)
-			go copyLayerHelper(i, srcLayer, layersToEncrypt.Contains(i), progressPool, ic.c.rawSource.Reference().DockerReference())
+			go copyLayerHelper(i, srcLayer, layersToEncrypt.Contains(i), progressPool, ic.c.rawSource.Reference().DockerReference(), deltaLayers)
 		}
 
 		// A call to copyGroup.Wait() is done at this point by the defer above.
@@ -690,10 +697,85 @@ func compressionEditsFromBlobInfo(srcInfo types.BlobInfo) (types.LayerCompressio
 	}
 }
 
+// getMatchingDeltaLayers gets all the deltas that apply to this layer
+func (ic *imageCopier) getMatchingDeltaLayers(ctx context.Context, srcIndex int, deltaLayers []types.BlobInfo) (digest.Digest, []*types.BlobInfo) {
+	if deltaLayers == nil {
+		return "", nil
+	}
+	config, _ := ic.src.OCIConfig(ctx)
+	if config == nil || config.RootFS.DiffIDs == nil || len(config.RootFS.DiffIDs) <= srcIndex {
+		return "", nil
+	}
+
+	layerDiffID := config.RootFS.DiffIDs[srcIndex]
+
+	var matchingLayers []*types.BlobInfo
+	for i := range deltaLayers {
+		deltaLayer := &deltaLayers[i]
+		to := deltaLayer.Annotations["io.github.containers.delta.to"]
+		if to == layerDiffID.String() {
+			matchingLayers = append(matchingLayers, deltaLayer)
+		}
+	}
+
+	return layerDiffID, matchingLayers
+}
+
+// resolveDeltaLayer looks at which of the matching delta froms have locally available data and picks the best one
+func (ic *imageCopier) resolveDeltaLayer(ctx context.Context, matchingDeltas []*types.BlobInfo) (io.ReadCloser, tarpatch.DataSource, types.BlobInfo, error) {
+	// Sort smallest deltas so we favour the smallest useable one
+	sort.Slice(matchingDeltas, func(i, j int) bool {
+		return matchingDeltas[i].Size < matchingDeltas[j].Size
+	})
+
+	for i := range matchingDeltas {
+		matchingDelta := matchingDeltas[i]
+		from := matchingDelta.Annotations["io.github.containers.delta.from"]
+		fromDigest, err := digest.Parse(from)
+		if err != nil {
+			continue // Silently ignore if server specified a weird format
+		}
+
+		dataSource, err := types.ImageDestinationGetLayerDeltaData(ic.c.dest, ctx, fromDigest)
+		if err != nil {
+			return nil, nil, types.BlobInfo{}, err // Internal error
+		}
+		if dataSource == nil {
+			continue // from layer doesn't exist
+		}
+
+		logrus.Debugf("Using delta %v for DiffID %v", matchingDelta.Digest, fromDigest)
+
+		deltaStream, _, err := ic.c.rawSource.GetBlob(ctx, *matchingDelta, ic.c.blobInfoCache)
+		if err != nil {
+			return nil, nil, types.BlobInfo{}, fmt.Errorf("reading delta blob %s: %w", matchingDelta.Digest, err)
+		}
+		return deltaStream, dataSource, *matchingDelta, nil
+	}
+	return nil, nil, types.BlobInfo{}, nil
+}
+
+// canUseDeltas checks if deltas can be used for this layer
+func (ic *imageCopier) canUseDeltas(srcInfo types.BlobInfo) (bool, string) {
+	// Deltas rewrite the manifest to refer to the uncompressed digest, so we must be able to substitute blobs
+	if !ic.canSubstituteBlobs {
+		return false, ""
+	}
+
+	switch srcInfo.MediaType {
+	case manifest.DockerV2Schema2LayerMediaType, manifest.DockerV2SchemaLayerMediaTypeUncompressed:
+		return true, manifest.DockerV2SchemaLayerMediaTypeUncompressed
+	case imgspecv1.MediaTypeImageLayer, imgspecv1.MediaTypeImageLayerGzip, imgspecv1.MediaTypeImageLayerZstd:
+		return true, imgspecv1.MediaTypeImageLayer
+	}
+
+	return false, ""
+}
+
 // copyLayer copies a layer with srcInfo (with known Digest and Annotations and possibly known Size) in src to dest, perhaps (de/re/)compressing it,
 // and returns a complete blobInfo of the copied layer, and a value for LayerDiffIDs if diffIDIsNeeded
 // srcRef can be used as an additional hint to the destination during checking whether a layer can be reused but srcRef can be nil.
-func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, toEncrypt bool, pool *mpb.Progress, layerIndex int, srcRef reference.Named, emptyLayer bool) (types.BlobInfo, digest.Digest, error) {
+func (ic *imageCopier) copyLayer(ctx context.Context, srcIndex int, srcInfo types.BlobInfo, toEncrypt bool, pool *mpb.Progress, srcRef reference.Named, emptyLayer bool, deltaLayers []types.BlobInfo) (types.BlobInfo, digest.Digest, error) {
 	// If the srcInfo doesn't contain compression information, try to compute it from the
 	// MediaType, which was either read from a manifest by way of LayerInfos() or constructed
 	// by LayerInfosForCopy(), if it was supplied at all.  If we succeed in copying the blob,
@@ -712,6 +794,58 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 
 	ic.c.printCopyInfo("blob", srcInfo)
 
+	// First look for a delta that matches this layer and substitute the result of that
+	if ok, deltaResultMediaType := ic.canUseDeltas(srcInfo); ok {
+		// Get deltas going TO this layer
+		deltaDiffID, matchingDeltas := ic.getMatchingDeltaLayers(ctx, srcIndex, deltaLayers)
+		// Get best possible FROM delta
+		deltaStream, deltaDataSource, matchingDelta, err := ic.resolveDeltaLayer(ctx, matchingDeltas)
+		if err != nil {
+			return types.BlobInfo{}, "", err
+		}
+		if deltaStream != nil {
+			bar, err := ic.c.createProgressBar(pool, false, matchingDelta, "delta", "done")
+			if err != nil {
+				return types.BlobInfo{}, "", err
+			}
+
+			wrappedDeltaStream := bar.ProxyReader(deltaStream)
+
+			// Convert deltaStream to uncompressed tar layer stream
+			pr, pw := io.Pipe()
+			go func() {
+				if err := tarpatch.Apply(wrappedDeltaStream, deltaDataSource, pw); err != nil {
+					// We will notice this error when failing to verify the digest, so leave it be
+					logrus.Infof("Failed to apply layer delta: %v", err)
+				}
+				deltaDataSource.Close()
+				deltaStream.Close()
+				wrappedDeltaStream.Close()
+				pw.Close()
+			}()
+			defer pr.Close()
+
+			// Copy uncompressed tar layer to destination, verifying the diffID
+			blobInfo, diffIDChan, err := ic.copyLayerFromStream(ctx, pr, types.BlobInfo{Digest: deltaDiffID, Size: -1, MediaType: deltaResultMediaType, Annotations: srcInfo.Annotations}, true, toEncrypt, bar, srcIndex, emptyLayer)
+			if err != nil {
+				return types.BlobInfo{}, "", err
+			}
+
+			// Wait for diffID verification
+			diffIDResult := <-diffIDChan
+			if diffIDResult.err != nil {
+				return types.BlobInfo{}, "", diffIDResult.err
+			}
+
+			bar.SetTotal(matchingDelta.Size, true)
+
+			// Record the fact that this blob is uncompressed
+			ic.c.blobInfoCache.RecordDigestUncompressedPair(diffIDResult.digest, diffIDResult.digest)
+
+			return blobInfo, diffIDResult.digest, nil
+		}
+	}
+
 	diffIDIsNeeded := false
 	var cachedDiffID digest.Digest = ""
 	if ic.diffIDsAreNeeded {
@@ -751,7 +885,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 			Cache:                   ic.c.blobInfoCache,
 			CanSubstitute:           canSubstitute,
 			EmptyLayer:              emptyLayer,
-			LayerIndex:              &layerIndex,
+			LayerIndex:              &srcIndex,
 			SrcRef:                  srcRef,
 			PossibleManifestFormats: append([]string{ic.manifestConversionPlan.preferredMIMEType}, ic.manifestConversionPlan.otherMIMETypeCandidates...),
 			RequiredCompression:     requiredCompression,
@@ -813,7 +947,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 			uploadedBlob, err := ic.c.dest.PutBlobPartial(ctx, &proxy, srcInfo, private.PutBlobPartialOptions{
 				Cache:      ic.c.blobInfoCache,
 				EmptyLayer: emptyLayer,
-				LayerIndex: layerIndex,
+				LayerIndex: srcIndex,
 			})
 			if err == nil {
 				if srcInfo.Size != -1 {
@@ -856,7 +990,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 		}
 		defer srcStream.Close()
 
-		blobInfo, diffIDChan, err := ic.copyLayerFromStream(ctx, srcStream, types.BlobInfo{Digest: srcInfo.Digest, Size: srcBlobSize, MediaType: srcInfo.MediaType, Annotations: srcInfo.Annotations}, diffIDIsNeeded, toEncrypt, bar, layerIndex, emptyLayer)
+		blobInfo, diffIDChan, err := ic.copyLayerFromStream(ctx, srcStream, types.BlobInfo{Digest: srcInfo.Digest, Size: srcBlobSize, MediaType: srcInfo.MediaType, Annotations: srcInfo.Annotations}, diffIDIsNeeded, toEncrypt, bar, srcIndex, emptyLayer)
 		if err != nil {
 			return types.BlobInfo{}, "", err
 		}

image/docker/docker_image_src.go

@@ -657,6 +657,47 @@ func (s *dockerImageSource) appendSignaturesFromSigstoreAttachments(ctx context.
 	return nil
 }
 
+// GetDeltaManifest returns the delta manifest for the current image, as well as its type, if it exists.
+// No error is returned if no delta manifest exists, just a nil slice
+func (s *dockerImageSource) GetDeltaManifest(ctx context.Context, instanceDigest *digest.Digest) ([]byte, string, error) {
+	// Get the real manifest digest
+	srcManifestDigest, err := s.manifestDigest(ctx, instanceDigest)
+	if err != nil {
+		return nil, "", err
+	}
+
+	// Load the delta manifest index
+	ib, _, err := s.fetchManifest(ctx, "_deltaindex")
+	// Don't return error if the manifest doesn't exist, only for internal errors
+	// Deltas are an optional optimization anyway
+	if err == nil {
+		index, err := manifest.OCI1IndexFromManifest(ib)
+		if err != nil {
+			return nil, "", err
+		}
+
+		// Look up the delta manifest in the index by the real manifest digest
+		for _, m := range index.Manifests {
+			if m.Annotations["io.github.containers.delta.target"] == srcManifestDigest.String() {
+				return s.fetchManifest(ctx, m.Digest.String())
+			}
+		}
+	}
+
+	// No delta
+	return nil, "", nil
+}
+
+// GetDeltaIndex returns an ImageReference that can be used to update the delta index for deltas for Image.
+func (s *dockerImageSource) GetDeltaIndex(ctx context.Context) (types.ImageReference, error) {
+	deltaRef, err := reference.WithTag(s.logicalRef.ref, "_deltaindex")
+	if err != nil {
+		return nil, err
+	}
+
+	return newReference(deltaRef, false)
+}
+
 // deleteImage deletes the named image from the registry, if supported.
 func deleteImage(ctx context.Context, sys *types.SystemContext, ref dockerReference) error {
 	if ref.isUnknownDigest {

image/go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/BurntSushi/toml v1.5.0
 	github.com/containers/libtrust v0.0.0-20230121012942-c1716e8a8d01
 	github.com/containers/ocicrypt v1.2.1
+	github.com/containers/tar-diff v0.1.2
 	github.com/cyberphone/json-canonicalization v0.0.0-20241213102144-19d51d7fe467
 	github.com/distribution/reference v0.6.0
 	github.com/docker/cli v29.0.4+incompatible

image/internal/image/unparsed.go

@@ -123,3 +123,23 @@ func (i *UnparsedImage) UntrustedSignatures(ctx context.Context) ([]signature.Si
 	}
 	return i.cachedSignatures, nil
 }
+
+// DeltaLayers downloads and parses the delta manifest for the image, returning the available delta layers
+func (i *UnparsedImage) DeltaLayers(ctx context.Context) ([]types.BlobInfo, error) {
+	// Note that GetDeltaManifest can return nil with a nil error. This is ok if no deltas exist
+	mb, mt, err := types.ImageSourceGetDeltaManifest(i.src, ctx, i.instanceDigest)
+	if mb == nil {
+		return nil, err
+	}
+
+	m, err := manifest.FromBlob(mb, mt)
+	if err != nil {
+		return nil, err
+	}
+	layerInfos := m.LayerInfos()
+	blobInfos := make([]types.BlobInfo, len(layerInfos))
+	for i, li := range layerInfos {
+		blobInfos[i] = li.BlobInfo
+	}
+	return blobInfos, nil
+}

image/internal/testing/mocks/image_source.go

@@ -45,3 +45,13 @@ func (f ForbiddenImageSource) GetSignatures(context.Context, *digest.Digest) ([]
 func (f ForbiddenImageSource) LayerInfosForCopy(context.Context, *digest.Digest) ([]types.BlobInfo, error) {
 	panic("Unexpected call to a mock function")
 }
+
+// GetDeltaManifest is a mock that panics.
+func (f ForbiddenImageSource) GetDeltaManifest(context.Context, *digest.Digest) ([]byte, string, error) {
+	panic("Unexpected call to a mock function")
+}
+
+// GetDeltaIndex is a mock that panics.
+func (f ForbiddenImageSource) GetDeltaIndex(context.Context) (types.ImageReference, error) {
+	panic("Unexpected call to a mock function")
+}

image/internal/testing/mocks/unparsed_image.go

@@ -29,3 +29,8 @@ func (ref ForbiddenUnparsedImage) Signatures(context.Context) ([][]byte, error)
 func (ref ForbiddenUnparsedImage) UntrustedSignatures(ctx context.Context) ([]signature.Signature, error) {
 	panic("unexpected call to a mock function")
 }
+
+// DeltaLayers is a mock that panics.
+func (ref ForbiddenUnparsedImage) DeltaLayers(ctx context.Context) ([]types.BlobInfo, error) {
+	panic("unexpected call to a mock function")
+}

image/manifest/oci.go

@@ -26,6 +26,11 @@ func BlobInfoFromOCI1Descriptor(desc imgspecv1.Descriptor) types.BlobInfo {
 	}
 }
 
+const (
+	// MediaTypeTarDiff specifies the media type for a tar-diff delta layer.
+	MediaTypeTarDiff = "application/vnd.tar-diff"
+)
+
 // OCI1 is a manifest.Manifest implementation for OCI images.
 // The underlying data from imgspecv1.Manifest is also available.
 type OCI1 struct {
@@ -47,7 +52,8 @@ func SupportedOCI1MediaType(m string) error {
 		imgspecv1.MediaTypeImageLayerNonDistributable, imgspecv1.MediaTypeImageLayerNonDistributableGzip, imgspecv1.MediaTypeImageLayerNonDistributableZstd, //nolint:staticcheck // NonDistributable layers are deprecated, but we want to continue to support manipulating pre-existing images.
 		imgspecv1.MediaTypeImageManifest,
 		imgspecv1.MediaTypeLayoutHeader,
-		ociencspec.MediaTypeLayerEnc, ociencspec.MediaTypeLayerGzipEnc:
+		ociencspec.MediaTypeLayerEnc, ociencspec.MediaTypeLayerGzipEnc,
+		MediaTypeTarDiff:
 		return nil
 	default:
 		return fmt.Errorf("unsupported OCIv1 media type: %q", m)
@@ -274,3 +280,8 @@ func (m *OCI1) CanChangeLayerCompression(mimeType string) bool {
 	}
 	return compressionVariantsRecognizeMIMEType(oci1CompressionMIMETypeSets, mimeType)
 }
+
+// IsNoCompressType returns true if the media type should not be recompressed
+func IsNoCompressType(mediatype string) bool {
+	return mediatype == MediaTypeTarDiff
+}

image/openshift/openshift_dest.go

@@ -252,3 +252,8 @@ func (d *openshiftImageDestination) PutSignaturesWithFormat(ctx context.Context,
 func (d *openshiftImageDestination) CommitWithOptions(ctx context.Context, options private.CommitOptions) error {
 	return d.docker.CommitWithOptions(ctx, options)
 }
+
+// GetLayerDeltaData provides access to local layer data by DiffID for delta application
+func (d *openshiftImageDestination) GetLayerDeltaData(ctx context.Context, diffID digest.Digest) (types.DeltaDataSource, error) {
+	return types.ImageDestinationGetLayerDeltaData(d.docker, ctx, diffID)
+}