feat(earthlyoutputs): serve original content of digest-pinned images

exporter/earthlyoutputs/export.go

@@ -28,12 +28,14 @@ import (
 	"github.com/moby/buildkit/session/filesync"
 	"github.com/moby/buildkit/session/pullping"
 	"github.com/moby/buildkit/snapshot"
+	"github.com/moby/buildkit/util/bklog"
 	"github.com/moby/buildkit/util/compression"
 	"github.com/moby/buildkit/util/contentutil"
 	"github.com/moby/buildkit/util/grpcerrors"
 	"github.com/moby/buildkit/util/leaseutil"
 	"github.com/moby/buildkit/util/progress"
 	"github.com/moby/buildkit/util/push"
+	"github.com/moby/buildkit/util/resolver"
 	digest "github.com/opencontainers/go-digest"
 	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
@@ -203,6 +205,9 @@ type imgData struct {
 	// localRegExport is set when the image should be exported via local registry. The value
 	// represents the image name that can be used to pull the image from the local registry.
 	localRegExport string
+	// localRegSource is the digest-pinned image a local registry export was pulled from,
+	// if any. See keyLocalRegistrySource.
+	localRegSource reference.Canonical
 
 	// shouldPush is set when the image should be pushed.
 	shouldPush bool
@@ -288,6 +293,13 @@ func (e *imageExporterInstance) Export(ctx context.Context, src *exporter.Source
 			eilr = string(simpleMd["export-image-local-registry"])
 			hasAnyLocalRegExport = true
 		}
+		var eilrSrc reference.Canonical
+		if v := string(simpleMd[keyLocalRegistrySource]); v != "" && eilr != "" {
+			eilrSrc, err = parseSourceRef(v)
+			if err != nil {
+				return nil, nil, err
+			}
+		}
 		sp := false
 		if string(simpleMd["export-image-push"]) == "true" {
 			isImage = true
@@ -332,6 +344,7 @@ func (e *imageExporterInstance) Export(ctx context.Context, src *exporter.Source
 				}
 				img.localExport = img.localExport || le
 				img.localRegExport = eilr
+				img.localRegSource = eilrSrc
 				img.shouldPush = img.shouldPush || sp
 				img.insecurePush = img.insecurePush || ip
 				img.opts = opts
@@ -550,6 +563,10 @@ func (e *imageExporterInstance) Export(ctx context.Context, src *exporter.Source
 			}
 		}
 
+		if img.localRegSource != nil {
+			e.addLocalRegSource(ctx, sessionID, img)
+		}
+
 		if img.shouldPush {
 			err := push.Push(
 				ctx, e.opt.SessionManager, sessionID, img.mp,
@@ -641,6 +658,35 @@ func (e *imageExporterInstance) Export(ctx context.Context, src *exporter.Source
 	return resp, nil, nil
 }
 
+// addLocalRegSource serves the original content of img.localRegSource from the
+// embedded registry as part of img. This is best effort: on failure, clients
+// pulling the source reference fall back to the upstream registry.
+func (e *imageExporterInstance) addLocalRegSource(ctx context.Context, sessionID string, img *imgData) {
+	cs := e.opt.ImageWriter.ContentStore()
+	err := func() error {
+		platform, err := imagePlatform(ctx, cs, *img.mfstDesc)
+		if err != nil {
+			return err
+		}
+		ref := img.localRegSource.String()
+		rslvr := resolver.DefaultPool.GetResolver(e.opt.RegistryHosts, ref, "pull", e.opt.SessionManager, session.NewGroup(sessionID))
+		descs, err := sourceContent(ctx, cs, rslvr, img.localRegSource, platform)
+		if err != nil {
+			return err
+		}
+		mmp := eodriver.MultiMultiProviderSingleton
+		for _, desc := range descs {
+			if err := mmp.AddImgSub(img.localRegExport, desc.Digest, cs); err != nil {
+				return err
+			}
+		}
+		return nil
+	}()
+	if err != nil {
+		bklog.G(ctx).WithError(err).Warnf("not serving original content of %s from the local registry", img.localRegSource)
+	}
+}
+
 func (e *imageExporterInstance) Config() *exporter.Config {
 	return exporter.NewConfigWithCompression(e.opts.RefCfg.Compression)
 }

exporter/earthlyoutputs/registry/eodriver/eodriver.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/containerd/containerd/content"
+	"github.com/containerd/containerd/errdefs"
 	storagedriver "github.com/docker/distribution/registry/storage/driver"
 	"github.com/docker/distribution/registry/storage/driver/base"
 	"github.com/docker/distribution/registry/storage/driver/factory"
@@ -166,7 +167,7 @@ func (d *driver) get(ctx context.Context, path string, offset int64) (io.ReadClo
 				fullImgName := fmt.Sprintf("%s:%s", imgName, tag)
 				_, baseDigest, err := d.mmp.Get(ctx, fullImgName)
 				if err != nil {
-					return nil, 0, errors.Wrapf(err, "get %s", path)
+					return nil, 0, notFoundOr(err, path)
 				}
 				return stringReadCloserOffset(baseDigest.String(), offset)
 			case "revisions":
@@ -204,7 +205,7 @@ func (d *driver) get(ctx context.Context, path string, offset int64) (io.ReadClo
 		}
 		ra, err := d.mmp.ReaderAt(ctx, desc)
 		if err != nil {
-			return nil, 0, errors.Wrapf(err, "blob %s", path)
+			return nil, 0, notFoundOr(err, path)
 		}
 		return &readerAtReadCloser{
 			ra:     ra,
@@ -294,6 +295,17 @@ func (fi fileInfo) IsDir() bool {
 	return false
 }
 
+// notFoundOr maps a not found error to storagedriver.PathNotFoundError, which
+// the registry reports as an unknown manifest or blob (404) rather than an
+// internal error (500). This matters when the registry is used as a mirror:
+// most requests are for content it does not have.
+func notFoundOr(err error, path string) error {
+	if errdefs.IsNotFound(err) {
+		return storagedriver.PathNotFoundError{Path: path, DriverName: driverName}
+	}
+	return errors.Wrapf(err, "get %s", path)
+}
+
 func stringReadCloserOffset(str string, offset int64) (io.ReadCloser, int64, error) {
 	dt := []byte(str)
 	if offset > int64(len(dt)) {

exporter/earthlyoutputs/registry/eodriver/eodriver_test.go

@@ -0,0 +1,25 @@
+package eodriver
+
+import (
+	"context"
+	"testing"
+
+	storagedriver "github.com/docker/distribution/registry/storage/driver"
+	digest "github.com/opencontainers/go-digest"
+	"github.com/stretchr/testify/require"
+)
+
+// Unknown content must be reported as not found, so that the registry answers
+// 404 rather than 500, e.g. when used as a mirror.
+func TestGetUnknownIsPathNotFound(t *testing.T) {
+	d := &driver{mmp: NewMultiMultiProvider()}
+	sha := digest.FromString("unknown").Encoded()
+
+	for _, path := range []string{
+		"/docker/registry/v2/repositories/library/alpine/_manifests/tags/latest/current/link",
+		"/docker/registry/v2/blobs/sha256/" + sha[:2] + "/" + sha + "/data",
+	} {
+		_, _, err := d.get(context.Background(), path, 0)
+		require.ErrorAs(t, err, &storagedriver.PathNotFoundError{}, path)
+	}
+}

exporter/earthlyoutputs/source.go

@@ -0,0 +1,124 @@
+package earthlyoutputs
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+
+	"github.com/containerd/containerd/content"
+	cerrdefs "github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/platforms"
+	"github.com/containerd/containerd/remotes"
+	"github.com/docker/distribution/reference"
+	"github.com/moby/buildkit/util/imageutil"
+	ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
+	"github.com/pkg/errors"
+)
+
+// keyLocalRegistrySource is the per-ref metadata key naming the image a local
+// registry export was pulled from, as a digest-pinned reference. When set, the
+// original index, manifest and config of that image are served from the
+// embedded registry alongside the exported image, so that a client can pull
+// the pinned reference itself (e.g. via a registry mirror) and have its digest
+// verify. Layers are not added: unmodified layers keep their original digests
+// and are already served as part of the exported image.
+const keyLocalRegistrySource = "export-image-local-registry-source"
+
+// parseSourceRef parses the value of keyLocalRegistrySource.
+func parseSourceRef(s string) (reference.Canonical, error) {
+	named, err := reference.ParseNormalizedNamed(s)
+	if err != nil {
+		return nil, errors.Wrapf(err, "parse %s %q", keyLocalRegistrySource, s)
+	}
+	canonical, ok := named.(reference.Canonical)
+	if !ok {
+		return nil, errors.Errorf("%s %q is not pinned by digest", keyLocalRegistrySource, s)
+	}
+	return canonical, nil
+}
+
+// sourceContent makes sure the non-layer content (index, the manifest matching
+// platform, and its config) of the image ref is in cs, and returns its
+// descriptors. Content already in cs is used as is: images pulled by BuildKit
+// keep this content leased to their cache records, so the common case needs no
+// network access. Anything missing is fetched with resolver.
+func sourceContent(ctx context.Context, cs content.Store, resolver remotes.Resolver, ref reference.Canonical, platform ocispecs.Platform) ([]ocispecs.Descriptor, error) {
+	root, err := rootDescriptor(ctx, cs, resolver, ref)
+	if err != nil {
+		return nil, err
+	}
+
+	fetcher, err := resolver.Fetcher(ctx, ref.String())
+	if err != nil {
+		return nil, errors.Wrapf(err, "fetcher for %s", ref)
+	}
+
+	matcher := platforms.Only(platform)
+	children := images.LimitManifests(images.FilterPlatforms(images.ChildrenHandler(cs), matcher), matcher, 1)
+
+	var (
+		mu    sync.Mutex
+		descs []ocispecs.Descriptor
+	)
+	collect := images.HandlerFunc(func(ctx context.Context, desc ocispecs.Descriptor) ([]ocispecs.Descriptor, error) {
+		if images.IsLayerType(desc.MediaType) {
+			return nil, images.ErrSkipDesc
+		}
+		mu.Lock()
+		descs = append(descs, desc)
+		mu.Unlock()
+		return nil, nil
+	})
+
+	// FetchHandler is a no-op for content that is already in cs.
+	if err := images.Dispatch(ctx, images.Handlers(collect, remotes.FetchHandler(cs, fetcher), children), nil, root); err != nil {
+		return nil, errors.Wrapf(err, "fetch %s", ref)
+	}
+	return descs, nil
+}
+
+// rootDescriptor returns the descriptor ref's digest points to, from cs if
+// present there, otherwise by resolving ref.
+func rootDescriptor(ctx context.Context, cs content.Store, resolver remotes.Resolver, ref reference.Canonical) (ocispecs.Descriptor, error) {
+	info, err := cs.Info(ctx, ref.Digest())
+	switch {
+	case err == nil:
+		desc := ocispecs.Descriptor{Digest: info.Digest, Size: info.Size}
+		dt, err := content.ReadBlob(ctx, cs, desc)
+		if err != nil {
+			return ocispecs.Descriptor{}, errors.Wrapf(err, "read %s", ref)
+		}
+		desc.MediaType, err = imageutil.DetectManifestBlobMediaType(dt)
+		if err != nil {
+			return ocispecs.Descriptor{}, errors.Wrapf(err, "detect media type of %s", ref)
+		}
+		return desc, nil
+	case cerrdefs.IsNotFound(err):
+		_, desc, err := resolver.Resolve(ctx, ref.String())
+		if err != nil {
+			return ocispecs.Descriptor{}, errors.Wrapf(err, "resolve %s", ref)
+		}
+		return desc, nil
+	default:
+		return ocispecs.Descriptor{}, errors.Wrapf(err, "stat %s", ref)
+	}
+}
+
+// imagePlatform returns the platform of the image described by the manifest
+// desc, as recorded in its config.
+func imagePlatform(ctx context.Context, provider content.Provider, desc ocispecs.Descriptor) (ocispecs.Platform, error) {
+	mfst, err := images.Manifest(ctx, provider, desc, platforms.All)
+	if err != nil {
+		return ocispecs.Platform{}, errors.Wrap(err, "read manifest")
+	}
+	dt, err := content.ReadBlob(ctx, provider, mfst.Config)
+	if err != nil {
+		return ocispecs.Platform{}, errors.Wrap(err, "read config")
+	}
+	var img ocispecs.Image
+	if err := json.Unmarshal(dt, &img); err != nil {
+		return ocispecs.Platform{}, errors.Wrap(err, "unmarshal config")
+	}
+	return platforms.Normalize(img.Platform), nil
+}