mirror of
https://github.com/moby/buildkit.git
synced 2026-08-08 08:40:45 +00:00
split puller from source/containerimage into util/pull
Signed-off-by: Akihiro Suda <suda.akihiro@lab.ntt.co.jp>
This commit is contained in:
Akihiro Suda
@@ -5,29 +5,18 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"runtime"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/diff"
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/platforms"
|
||||
"github.com/containerd/containerd/remotes"
|
||||
"github.com/containerd/containerd/remotes/docker"
|
||||
"github.com/containerd/containerd/remotes/docker/schema1"
|
||||
"github.com/containerd/containerd/rootfs"
|
||||
"github.com/containerd/containerd/snapshots"
|
||||
"github.com/docker/distribution/reference"
|
||||
"github.com/moby/buildkit/cache"
|
||||
"github.com/moby/buildkit/session"
|
||||
"github.com/moby/buildkit/session/auth"
|
||||
"github.com/moby/buildkit/snapshot"
|
||||
"github.com/moby/buildkit/source"
|
||||
"github.com/moby/buildkit/util/flightcontrol"
|
||||
"github.com/moby/buildkit/util/imageutil"
|
||||
"github.com/moby/buildkit/util/progress"
|
||||
"github.com/moby/buildkit/util/tracing"
|
||||
"github.com/moby/buildkit/util/pull"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
"github.com/opencontainers/image-spec/identity"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
@@ -63,44 +52,13 @@ func (is *imageSource) ID() string {
|
||||
return source.DockerImageScheme
|
||||
}
|
||||
|
||||
func (is *imageSource) getResolver(ctx context.Context) remotes.Resolver {
|
||||
r := docker.NewResolver(docker.ResolverOptions{
|
||||
Client: tracing.DefaultClient,
|
||||
Credentials: is.getCredentialsFromSession(ctx),
|
||||
})
|
||||
|
||||
if is.ImageStore == nil {
|
||||
return r
|
||||
}
|
||||
|
||||
return localFallbackResolver{r, is.ImageStore}
|
||||
}
|
||||
|
||||
func (is *imageSource) getCredentialsFromSession(ctx context.Context) func(string) (string, string, error) {
|
||||
id := session.FromContext(ctx)
|
||||
if id == "" {
|
||||
return nil
|
||||
}
|
||||
return func(host string) (string, string, error) {
|
||||
timeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
caller, err := is.SessionManager.Get(timeoutCtx, id)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
return auth.CredentialsFunc(tracing.ContextWithSpanFromContext(context.TODO(), ctx), caller)(host)
|
||||
}
|
||||
}
|
||||
|
||||
func (is *imageSource) ResolveImageConfig(ctx context.Context, ref string) (digest.Digest, []byte, error) {
|
||||
type t struct {
|
||||
dgst digest.Digest
|
||||
dt []byte
|
||||
}
|
||||
res, err := is.g.Do(ctx, ref, func(ctx context.Context) (interface{}, error) {
|
||||
dgst, dt, err := imageutil.Config(ctx, ref, is.getResolver(ctx), is.ContentStore)
|
||||
dgst, dt, err := imageutil.Config(ctx, ref, pull.NewResolver(ctx, is.SessionManager, is.ImageStore), is.ContentStore)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -119,102 +77,32 @@ func (is *imageSource) Resolve(ctx context.Context, id source.Identifier) (sourc
|
||||
return nil, errors.Errorf("invalid image identifier %v", id)
|
||||
}
|
||||
|
||||
pullerUtil := &pull.Puller{
|
||||
Snapshotter: is.Snapshotter,
|
||||
ContentStore: is.ContentStore,
|
||||
Applier: is.Applier,
|
||||
Src: imageIdentifier.Reference,
|
||||
Resolver: pull.NewResolver(ctx, is.SessionManager, is.ImageStore),
|
||||
}
|
||||
p := &puller{
|
||||
src: imageIdentifier,
|
||||
is: is,
|
||||
resolver: is.getResolver(ctx),
|
||||
CacheAccessor: is.CacheAccessor,
|
||||
Puller: pullerUtil,
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// A remotes.Resolver which checks the local image store if the real
|
||||
// resolver cannot find the image, essentially falling back to a local
|
||||
// image if one is present.
|
||||
//
|
||||
// We do not override the Fetcher or Pusher methods:
|
||||
//
|
||||
// - Fetcher is called by github.com/containerd/containerd/remotes/:fetch()
|
||||
// only after it has checked for the content locally, so avoid the
|
||||
// hassle of interposing a local-fetch proxy and simply pass on the
|
||||
// request.
|
||||
// - Pusher wouldn't make sense to push locally, so just forward.
|
||||
|
||||
type localFallbackResolver struct {
|
||||
remotes.Resolver
|
||||
is images.Store
|
||||
}
|
||||
|
||||
func (r localFallbackResolver) Resolve(ctx context.Context, ref string) (string, ocispec.Descriptor, error) {
|
||||
n, desc, err := r.Resolver.Resolve(ctx, ref)
|
||||
if err == nil {
|
||||
return n, desc, err
|
||||
}
|
||||
|
||||
img, err2 := r.is.Get(ctx, ref)
|
||||
if err2 != nil {
|
||||
return "", ocispec.Descriptor{}, err
|
||||
}
|
||||
return ref, img.Target, nil
|
||||
}
|
||||
|
||||
type puller struct {
|
||||
is *imageSource
|
||||
resolveOnce sync.Once
|
||||
src *source.ImageIdentifier
|
||||
desc ocispec.Descriptor
|
||||
ref string
|
||||
resolveErr error
|
||||
resolver remotes.Resolver
|
||||
CacheAccessor cache.Accessor
|
||||
*pull.Puller
|
||||
}
|
||||
|
||||
func (p *puller) resolve(ctx context.Context) error {
|
||||
p.resolveOnce.Do(func() {
|
||||
resolveProgressDone := oneOffProgress(ctx, "resolve "+p.src.Reference.String())
|
||||
|
||||
dgst := p.src.Reference.Digest()
|
||||
if dgst != "" {
|
||||
info, err := p.is.ContentStore.Info(ctx, dgst)
|
||||
if err == nil {
|
||||
p.ref = p.src.Reference.String()
|
||||
ra, err := p.is.ContentStore.ReaderAt(ctx, dgst)
|
||||
if err == nil {
|
||||
mt, err := imageutil.DetectManifestMediaType(ra)
|
||||
if err == nil {
|
||||
p.desc = ocispec.Descriptor{
|
||||
Size: info.Size,
|
||||
Digest: dgst,
|
||||
MediaType: mt,
|
||||
}
|
||||
resolveProgressDone(nil)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ref, desc, err := p.resolver.Resolve(ctx, p.src.Reference.String())
|
||||
if err != nil {
|
||||
p.resolveErr = err
|
||||
resolveProgressDone(err)
|
||||
return
|
||||
}
|
||||
p.desc = desc
|
||||
p.ref = ref
|
||||
resolveProgressDone(nil)
|
||||
})
|
||||
return p.resolveErr
|
||||
}
|
||||
|
||||
func (p *puller) mainManifestKey(ctx context.Context) (digest.Digest, error) {
|
||||
if err := p.resolve(ctx); err != nil {
|
||||
return "", err
|
||||
}
|
||||
func mainManifestKey(ctx context.Context, desc ocispec.Descriptor) (digest.Digest, error) {
|
||||
dt, err := json.Marshal(struct {
|
||||
Digest digest.Digest
|
||||
OS string
|
||||
Arch string
|
||||
}{
|
||||
Digest: p.desc.Digest,
|
||||
Digest: desc.Digest,
|
||||
OS: runtime.GOOS,
|
||||
Arch: runtime.GOARCH,
|
||||
})
|
||||
@@ -225,25 +113,29 @@ func (p *puller) mainManifestKey(ctx context.Context) (digest.Digest, error) {
|
||||
}
|
||||
|
||||
func (p *puller) CacheKey(ctx context.Context, index int) (string, bool, error) {
|
||||
if index == 0 || p.desc.Digest == "" {
|
||||
k, err := p.mainManifestKey(ctx)
|
||||
_, desc, err := p.Puller.Resolve(ctx)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
if index == 0 || desc.Digest == "" {
|
||||
k, err := mainManifestKey(ctx, desc)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
return k.String(), false, nil
|
||||
}
|
||||
ref, err := reference.ParseNormalizedNamed(p.src.Reference.String())
|
||||
ref, err := reference.ParseNormalizedNamed(p.Src.String())
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
ref, err = reference.WithDigest(ref, p.desc.Digest)
|
||||
ref, err = reference.WithDigest(ref, desc.Digest)
|
||||
if err != nil {
|
||||
return "", false, nil
|
||||
}
|
||||
_, dt, err := imageutil.Config(ctx, ref.String(), p.resolver, p.is.ContentStore)
|
||||
_, dt, err := imageutil.Config(ctx, ref.String(), p.Resolver, p.ContentStore)
|
||||
if err != nil {
|
||||
// this happens on schema1 images
|
||||
k, err := p.mainManifestKey(ctx)
|
||||
k, err := mainManifestKey(ctx, desc)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
@@ -253,300 +145,11 @@ func (p *puller) CacheKey(ctx context.Context, index int) (string, bool, error)
|
||||
}
|
||||
|
||||
func (p *puller) Snapshot(ctx context.Context) (cache.ImmutableRef, error) {
|
||||
if err := p.resolve(ctx); err != nil {
|
||||
pulled, err := p.Puller.Pull(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ongoing := newJobs(p.ref)
|
||||
|
||||
pctx, stopProgress := context.WithCancel(ctx)
|
||||
|
||||
go showProgress(pctx, ongoing, p.is.ContentStore)
|
||||
|
||||
fetcher, err := p.resolver.Fetcher(ctx, p.ref)
|
||||
if err != nil {
|
||||
stopProgress()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TODO: need a wrapper snapshot interface that combines content
|
||||
// and snapshots as 1) buildkit shouldn't have a dependency on contentstore
|
||||
// or 2) cachemanager should manage the contentstore
|
||||
handlers := []images.Handler{
|
||||
images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
ongoing.add(desc)
|
||||
return nil, nil
|
||||
}),
|
||||
}
|
||||
var schema1Converter *schema1.Converter
|
||||
if p.desc.MediaType == images.MediaTypeDockerSchema1Manifest {
|
||||
schema1Converter = schema1.NewConverter(p.is.ContentStore, fetcher)
|
||||
handlers = append(handlers, schema1Converter)
|
||||
} else {
|
||||
// Get all the children for a descriptor
|
||||
childrenHandler := images.ChildrenHandler(p.is.ContentStore)
|
||||
// Set any children labels for that content
|
||||
childrenHandler = images.SetChildrenLabels(p.is.ContentStore, childrenHandler)
|
||||
// Filter the childen by the platform
|
||||
childrenHandler = images.FilterPlatforms(childrenHandler, platforms.Default())
|
||||
|
||||
handlers = append(handlers,
|
||||
remotes.FetchHandler(p.is.ContentStore, fetcher),
|
||||
childrenHandler,
|
||||
)
|
||||
}
|
||||
|
||||
if err := images.Dispatch(ctx, images.Handlers(handlers...), p.desc); err != nil {
|
||||
stopProgress()
|
||||
return nil, err
|
||||
}
|
||||
stopProgress()
|
||||
|
||||
var usedBlobs, unusedBlobs []ocispec.Descriptor
|
||||
|
||||
if schema1Converter != nil {
|
||||
ongoing.remove(p.desc) // Not left in the content store so this is sufficient.
|
||||
p.desc, err = schema1Converter.Convert(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ongoing.add(p.desc)
|
||||
|
||||
var mu sync.Mutex // images.Dispatch calls handlers in parallel
|
||||
allBlobs := make(map[digest.Digest]ocispec.Descriptor)
|
||||
for _, j := range ongoing.added {
|
||||
allBlobs[j.Digest] = j.Descriptor
|
||||
}
|
||||
|
||||
handlers := []images.Handler{
|
||||
images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
usedBlobs = append(usedBlobs, desc)
|
||||
delete(allBlobs, desc.Digest)
|
||||
return nil, nil
|
||||
}),
|
||||
images.FilterPlatforms(images.ChildrenHandler(p.is.ContentStore), platforms.Default()),
|
||||
}
|
||||
|
||||
if err := images.Dispatch(ctx, images.Handlers(handlers...), p.desc); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, j := range allBlobs {
|
||||
unusedBlobs = append(unusedBlobs, j)
|
||||
}
|
||||
} else {
|
||||
for _, j := range ongoing.added {
|
||||
usedBlobs = append(usedBlobs, j.Descriptor)
|
||||
}
|
||||
}
|
||||
|
||||
// split all pulled data to layers and rest. layers remain roots and are deleted with snapshots. rest will be linked to layers.
|
||||
var notLayerBlobs []ocispec.Descriptor
|
||||
var layerBlobs []ocispec.Descriptor
|
||||
for _, j := range usedBlobs {
|
||||
switch j.MediaType {
|
||||
case ocispec.MediaTypeImageLayer, images.MediaTypeDockerSchema2Layer, ocispec.MediaTypeImageLayerGzip, images.MediaTypeDockerSchema2LayerGzip:
|
||||
layerBlobs = append(layerBlobs, j)
|
||||
default:
|
||||
notLayerBlobs = append(notLayerBlobs, j)
|
||||
}
|
||||
}
|
||||
|
||||
for _, l := range layerBlobs {
|
||||
labels := map[string]string{}
|
||||
var fields []string
|
||||
for _, nl := range notLayerBlobs {
|
||||
k := "containerd.io/gc.ref.content." + nl.Digest.Hex()[:12]
|
||||
labels[k] = nl.Digest.String()
|
||||
fields = append(fields, "labels."+k)
|
||||
}
|
||||
if _, err := p.is.ContentStore.Update(ctx, content.Info{
|
||||
Digest: l.Digest,
|
||||
Labels: labels,
|
||||
}, fields...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
for _, nl := range append(notLayerBlobs, unusedBlobs...) {
|
||||
if err := p.is.ContentStore.Delete(ctx, nl.Digest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
csh, release := snapshot.NewContainerdSnapshotter(p.is.Snapshotter)
|
||||
defer release()
|
||||
|
||||
unpackProgressDone := oneOffProgress(ctx, "unpacking "+p.src.Reference.String())
|
||||
chainid, err := p.is.unpack(ctx, p.desc, csh)
|
||||
if err != nil {
|
||||
return nil, unpackProgressDone(err)
|
||||
}
|
||||
unpackProgressDone(nil)
|
||||
|
||||
return p.is.CacheAccessor.GetFromSnapshotter(ctx, chainid, cache.WithDescription(fmt.Sprintf("pulled from %s", p.ref)))
|
||||
}
|
||||
|
||||
func (is *imageSource) unpack(ctx context.Context, desc ocispec.Descriptor, s snapshots.Snapshotter) (string, error) {
|
||||
layers, err := getLayers(ctx, is.ContentStore, desc)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var chain []digest.Digest
|
||||
for _, layer := range layers {
|
||||
labels := map[string]string{
|
||||
"containerd.io/gc.root": time.Now().UTC().Format(time.RFC3339Nano),
|
||||
"containerd.io/uncompressed": layer.Diff.Digest.String(),
|
||||
}
|
||||
if _, err := rootfs.ApplyLayer(ctx, layer, chain, s, is.Applier, snapshots.WithLabels(labels)); err != nil {
|
||||
return "", err
|
||||
}
|
||||
chain = append(chain, layer.Diff.Digest)
|
||||
}
|
||||
chainID := identity.ChainID(chain)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if err := is.fillBlobMapping(ctx, layers); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(chainID), nil
|
||||
}
|
||||
|
||||
func (is *imageSource) fillBlobMapping(ctx context.Context, layers []rootfs.Layer) error {
|
||||
var chain []digest.Digest
|
||||
for _, l := range layers {
|
||||
chain = append(chain, l.Diff.Digest)
|
||||
chainID := identity.ChainID(chain)
|
||||
if err := is.SourceOpt.Snapshotter.SetBlob(ctx, string(chainID), l.Diff.Digest, l.Blob.Digest); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func getLayers(ctx context.Context, provider content.Provider, desc ocispec.Descriptor) ([]rootfs.Layer, error) {
|
||||
manifest, err := images.Manifest(ctx, provider, desc, platforms.Default())
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
}
|
||||
image := images.Image{Target: desc}
|
||||
diffIDs, err := image.RootFS(ctx, provider, platforms.Default())
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to resolve rootfs")
|
||||
}
|
||||
if len(diffIDs) != len(manifest.Layers) {
|
||||
return nil, errors.Errorf("mismatched image rootfs and manifest layers %+v %+v", diffIDs, manifest.Layers)
|
||||
}
|
||||
layers := make([]rootfs.Layer, len(diffIDs))
|
||||
for i := range diffIDs {
|
||||
layers[i].Diff = ocispec.Descriptor{
|
||||
// TODO: derive media type from compressed type
|
||||
MediaType: ocispec.MediaTypeImageLayer,
|
||||
Digest: diffIDs[i],
|
||||
}
|
||||
layers[i].Blob = manifest.Layers[i]
|
||||
}
|
||||
return layers, nil
|
||||
}
|
||||
|
||||
func showProgress(ctx context.Context, ongoing *jobs, cs content.Store) {
|
||||
var (
|
||||
ticker = time.NewTicker(100 * time.Millisecond)
|
||||
statuses = map[string]statusInfo{}
|
||||
done bool
|
||||
)
|
||||
defer ticker.Stop()
|
||||
|
||||
pw, _, ctx := progress.FromContext(ctx)
|
||||
defer pw.Close()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-ctx.Done():
|
||||
done = true
|
||||
}
|
||||
|
||||
resolved := "resolved"
|
||||
if !ongoing.isResolved() {
|
||||
resolved = "resolving"
|
||||
}
|
||||
statuses[ongoing.name] = statusInfo{
|
||||
Ref: ongoing.name,
|
||||
Status: resolved,
|
||||
}
|
||||
|
||||
actives := make(map[string]statusInfo)
|
||||
|
||||
if !done {
|
||||
active, err := cs.ListStatuses(ctx, "")
|
||||
if err != nil {
|
||||
// log.G(ctx).WithError(err).Error("active check failed")
|
||||
continue
|
||||
}
|
||||
// update status of active entries!
|
||||
for _, active := range active {
|
||||
actives[active.Ref] = statusInfo{
|
||||
Ref: active.Ref,
|
||||
Status: "downloading",
|
||||
Offset: active.Offset,
|
||||
Total: active.Total,
|
||||
StartedAt: active.StartedAt,
|
||||
UpdatedAt: active.UpdatedAt,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// now, update the items in jobs that are not in active
|
||||
for _, j := range ongoing.jobs() {
|
||||
refKey := remotes.MakeRefKey(ctx, j.Descriptor)
|
||||
if a, ok := actives[refKey]; ok {
|
||||
started := j.started
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: a.Status,
|
||||
Total: int(a.Total),
|
||||
Current: int(a.Offset),
|
||||
Started: &started,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
if !j.done {
|
||||
info, err := cs.Info(context.TODO(), j.Digest)
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: "waiting",
|
||||
})
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
j.done = true
|
||||
}
|
||||
|
||||
if done || j.done {
|
||||
started := j.started
|
||||
createdAt := info.CreatedAt
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: "done",
|
||||
Current: int(info.Size),
|
||||
Total: int(info.Size),
|
||||
Completed: &createdAt,
|
||||
Started: &started,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
if done {
|
||||
return
|
||||
}
|
||||
}
|
||||
return p.CacheAccessor.GetFromSnapshotter(ctx, string(pulled.ChainID), cache.WithDescription(fmt.Sprintf("pulled from %s", pulled.Ref)))
|
||||
}
|
||||
|
||||
// cacheKeyFromConfig returns a stable digest from image config. If image config
|
||||
@@ -562,91 +165,3 @@ func cacheKeyFromConfig(dt []byte) digest.Digest {
|
||||
}
|
||||
return identity.ChainID(img.RootFS.DiffIDs)
|
||||
}
|
||||
|
||||
// jobs provides a way of identifying the download keys for a particular task
|
||||
// encountering during the pull walk.
|
||||
//
|
||||
// This is very minimal and will probably be replaced with something more
|
||||
// featured.
|
||||
type jobs struct {
|
||||
name string
|
||||
added map[digest.Digest]job
|
||||
mu sync.Mutex
|
||||
resolved bool
|
||||
}
|
||||
|
||||
type job struct {
|
||||
ocispec.Descriptor
|
||||
done bool
|
||||
started time.Time
|
||||
}
|
||||
|
||||
func newJobs(name string) *jobs {
|
||||
return &jobs{
|
||||
name: name,
|
||||
added: make(map[digest.Digest]job),
|
||||
}
|
||||
}
|
||||
|
||||
func (j *jobs) add(desc ocispec.Descriptor) {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
if _, ok := j.added[desc.Digest]; ok {
|
||||
return
|
||||
}
|
||||
j.added[desc.Digest] = job{
|
||||
Descriptor: desc,
|
||||
started: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
func (j *jobs) remove(desc ocispec.Descriptor) {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
delete(j.added, desc.Digest)
|
||||
}
|
||||
|
||||
func (j *jobs) jobs() []job {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
descs := make([]job, 0, len(j.added))
|
||||
for _, j := range j.added {
|
||||
descs = append(descs, j)
|
||||
}
|
||||
return descs
|
||||
}
|
||||
|
||||
func (j *jobs) isResolved() bool {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
return j.resolved
|
||||
}
|
||||
|
||||
type statusInfo struct {
|
||||
Ref string
|
||||
Status string
|
||||
Offset int64
|
||||
Total int64
|
||||
StartedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func oneOffProgress(ctx context.Context, id string) func(err error) error {
|
||||
pw, _, _ := progress.FromContext(ctx)
|
||||
now := time.Now()
|
||||
st := progress.Status{
|
||||
Started: &now,
|
||||
}
|
||||
pw.Write(id, st)
|
||||
return func(err error) error {
|
||||
// TODO: set error on status
|
||||
now := time.Now()
|
||||
st.Completed = &now
|
||||
pw.Write(id, st)
|
||||
pw.Close()
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
471
util/pull/pull.go
Normal file
471
util/pull/pull.go
Normal file
@@ -0,0 +1,471 @@
|
||||
package pull
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/diff"
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/platforms"
|
||||
"github.com/containerd/containerd/reference"
|
||||
"github.com/containerd/containerd/remotes"
|
||||
"github.com/containerd/containerd/remotes/docker/schema1"
|
||||
"github.com/containerd/containerd/rootfs"
|
||||
ctdsnapshot "github.com/containerd/containerd/snapshots"
|
||||
"github.com/moby/buildkit/snapshot"
|
||||
"github.com/moby/buildkit/util/imageutil"
|
||||
"github.com/moby/buildkit/util/progress"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
"github.com/opencontainers/image-spec/identity"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
type Puller struct {
|
||||
Snapshotter snapshot.Snapshotter
|
||||
ContentStore content.Store
|
||||
Applier diff.Applier
|
||||
Src reference.Spec
|
||||
// See NewResolver()
|
||||
Resolver remotes.Resolver
|
||||
resolveOnce sync.Once
|
||||
desc ocispec.Descriptor
|
||||
ref string
|
||||
resolveErr error
|
||||
}
|
||||
|
||||
type Pulled struct {
|
||||
Ref string
|
||||
Descriptor ocispec.Descriptor
|
||||
ChainID digest.Digest
|
||||
}
|
||||
|
||||
func (p *Puller) Resolve(ctx context.Context) (string, ocispec.Descriptor, error) {
|
||||
p.resolveOnce.Do(func() {
|
||||
resolveProgressDone := oneOffProgress(ctx, "resolve "+p.Src.String())
|
||||
|
||||
dgst := p.Src.Digest()
|
||||
if dgst != "" {
|
||||
info, err := p.ContentStore.Info(ctx, dgst)
|
||||
if err == nil {
|
||||
p.ref = p.Src.String()
|
||||
ra, err := p.ContentStore.ReaderAt(ctx, dgst)
|
||||
if err == nil {
|
||||
mt, err := imageutil.DetectManifestMediaType(ra)
|
||||
if err == nil {
|
||||
p.desc = ocispec.Descriptor{
|
||||
Size: info.Size,
|
||||
Digest: dgst,
|
||||
MediaType: mt,
|
||||
}
|
||||
resolveProgressDone(nil)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ref, desc, err := p.Resolver.Resolve(ctx, p.Src.String())
|
||||
if err != nil {
|
||||
p.resolveErr = err
|
||||
resolveProgressDone(err)
|
||||
return
|
||||
}
|
||||
p.desc = desc
|
||||
p.ref = ref
|
||||
resolveProgressDone(nil)
|
||||
})
|
||||
return p.ref, p.desc, p.resolveErr
|
||||
}
|
||||
|
||||
func (p *Puller) Pull(ctx context.Context) (*Pulled, error) {
|
||||
if _, _, err := p.Resolve(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ongoing := newJobs(p.ref)
|
||||
|
||||
pctx, stopProgress := context.WithCancel(ctx)
|
||||
|
||||
go showProgress(pctx, ongoing, p.ContentStore)
|
||||
|
||||
fetcher, err := p.Resolver.Fetcher(ctx, p.ref)
|
||||
if err != nil {
|
||||
stopProgress()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TODO: need a wrapper snapshot interface that combines content
|
||||
// and snapshots as 1) buildkit shouldn't have a dependency on contentstore
|
||||
// or 2) cachemanager should manage the contentstore
|
||||
handlers := []images.Handler{
|
||||
images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
ongoing.add(desc)
|
||||
return nil, nil
|
||||
}),
|
||||
}
|
||||
var schema1Converter *schema1.Converter
|
||||
if p.desc.MediaType == images.MediaTypeDockerSchema1Manifest {
|
||||
schema1Converter = schema1.NewConverter(p.ContentStore, fetcher)
|
||||
handlers = append(handlers, schema1Converter)
|
||||
} else {
|
||||
// Get all the children for a descriptor
|
||||
childrenHandler := images.ChildrenHandler(p.ContentStore)
|
||||
// Set any children labels for that content
|
||||
childrenHandler = images.SetChildrenLabels(p.ContentStore, childrenHandler)
|
||||
// Filter the childen by the platform
|
||||
childrenHandler = images.FilterPlatforms(childrenHandler, platforms.Default())
|
||||
|
||||
handlers = append(handlers,
|
||||
remotes.FetchHandler(p.ContentStore, fetcher),
|
||||
childrenHandler,
|
||||
)
|
||||
}
|
||||
|
||||
if err := images.Dispatch(ctx, images.Handlers(handlers...), p.desc); err != nil {
|
||||
stopProgress()
|
||||
return nil, err
|
||||
}
|
||||
stopProgress()
|
||||
|
||||
var usedBlobs, unusedBlobs []ocispec.Descriptor
|
||||
|
||||
if schema1Converter != nil {
|
||||
ongoing.remove(p.desc) // Not left in the content store so this is sufficient.
|
||||
p.desc, err = schema1Converter.Convert(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ongoing.add(p.desc)
|
||||
|
||||
var mu sync.Mutex // images.Dispatch calls handlers in parallel
|
||||
allBlobs := make(map[digest.Digest]ocispec.Descriptor)
|
||||
for _, j := range ongoing.added {
|
||||
allBlobs[j.Digest] = j.Descriptor
|
||||
}
|
||||
|
||||
handlers := []images.Handler{
|
||||
images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
usedBlobs = append(usedBlobs, desc)
|
||||
delete(allBlobs, desc.Digest)
|
||||
return nil, nil
|
||||
}),
|
||||
images.FilterPlatforms(images.ChildrenHandler(p.ContentStore), platforms.Default()),
|
||||
}
|
||||
|
||||
if err := images.Dispatch(ctx, images.Handlers(handlers...), p.desc); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, j := range allBlobs {
|
||||
unusedBlobs = append(unusedBlobs, j)
|
||||
}
|
||||
} else {
|
||||
for _, j := range ongoing.added {
|
||||
usedBlobs = append(usedBlobs, j.Descriptor)
|
||||
}
|
||||
}
|
||||
|
||||
// split all pulled data to layers and rest. layers remain roots and are deleted with snapshots. rest will be linked to layers.
|
||||
var notLayerBlobs []ocispec.Descriptor
|
||||
var layerBlobs []ocispec.Descriptor
|
||||
for _, j := range usedBlobs {
|
||||
switch j.MediaType {
|
||||
case ocispec.MediaTypeImageLayer, images.MediaTypeDockerSchema2Layer, ocispec.MediaTypeImageLayerGzip, images.MediaTypeDockerSchema2LayerGzip:
|
||||
layerBlobs = append(layerBlobs, j)
|
||||
default:
|
||||
notLayerBlobs = append(notLayerBlobs, j)
|
||||
}
|
||||
}
|
||||
|
||||
for _, l := range layerBlobs {
|
||||
labels := map[string]string{}
|
||||
var fields []string
|
||||
for _, nl := range notLayerBlobs {
|
||||
k := "containerd.io/gc.ref.content." + nl.Digest.Hex()[:12]
|
||||
labels[k] = nl.Digest.String()
|
||||
fields = append(fields, "labels."+k)
|
||||
}
|
||||
if _, err := p.ContentStore.Update(ctx, content.Info{
|
||||
Digest: l.Digest,
|
||||
Labels: labels,
|
||||
}, fields...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
for _, nl := range append(notLayerBlobs, unusedBlobs...) {
|
||||
if err := p.ContentStore.Delete(ctx, nl.Digest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
csh, release := snapshot.NewContainerdSnapshotter(p.Snapshotter)
|
||||
defer release()
|
||||
|
||||
unpackProgressDone := oneOffProgress(ctx, "unpacking "+p.Src.String())
|
||||
chainid, err := unpack(ctx, p.desc, p.ContentStore, csh, p.Snapshotter, p.Applier)
|
||||
if err != nil {
|
||||
return nil, unpackProgressDone(err)
|
||||
}
|
||||
unpackProgressDone(nil)
|
||||
|
||||
return &Pulled{
|
||||
Ref: p.ref,
|
||||
Descriptor: p.desc,
|
||||
ChainID: chainid,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func unpack(ctx context.Context, desc ocispec.Descriptor, cs content.Store, csh ctdsnapshot.Snapshotter, s snapshot.Snapshotter, applier diff.Applier) (digest.Digest, error) {
|
||||
layers, err := getLayers(ctx, cs, desc)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var chain []digest.Digest
|
||||
for _, layer := range layers {
|
||||
labels := map[string]string{
|
||||
"containerd.io/gc.root": time.Now().UTC().Format(time.RFC3339Nano),
|
||||
"containerd.io/uncompressed": layer.Diff.Digest.String(),
|
||||
}
|
||||
if _, err := rootfs.ApplyLayer(ctx, layer, chain, csh, applier, ctdsnapshot.WithLabels(labels)); err != nil {
|
||||
return "", err
|
||||
}
|
||||
chain = append(chain, layer.Diff.Digest)
|
||||
}
|
||||
chainID := identity.ChainID(chain)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if err := fillBlobMapping(ctx, s, layers); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return chainID, nil
|
||||
}
|
||||
|
||||
func fillBlobMapping(ctx context.Context, s snapshot.Snapshotter, layers []rootfs.Layer) error {
|
||||
var chain []digest.Digest
|
||||
for _, l := range layers {
|
||||
chain = append(chain, l.Diff.Digest)
|
||||
chainID := identity.ChainID(chain)
|
||||
if err := s.SetBlob(ctx, string(chainID), l.Diff.Digest, l.Blob.Digest); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func getLayers(ctx context.Context, provider content.Provider, desc ocispec.Descriptor) ([]rootfs.Layer, error) {
|
||||
manifest, err := images.Manifest(ctx, provider, desc, platforms.Default())
|
||||
if err != nil {
|
||||
return nil, errors.WithStack(err)
|
||||
}
|
||||
image := images.Image{Target: desc}
|
||||
diffIDs, err := image.RootFS(ctx, provider, platforms.Default())
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "failed to resolve rootfs")
|
||||
}
|
||||
if len(diffIDs) != len(manifest.Layers) {
|
||||
return nil, errors.Errorf("mismatched image rootfs and manifest layers %+v %+v", diffIDs, manifest.Layers)
|
||||
}
|
||||
layers := make([]rootfs.Layer, len(diffIDs))
|
||||
for i := range diffIDs {
|
||||
layers[i].Diff = ocispec.Descriptor{
|
||||
// TODO: derive media type from compressed type
|
||||
MediaType: ocispec.MediaTypeImageLayer,
|
||||
Digest: diffIDs[i],
|
||||
}
|
||||
layers[i].Blob = manifest.Layers[i]
|
||||
}
|
||||
return layers, nil
|
||||
}
|
||||
|
||||
func showProgress(ctx context.Context, ongoing *jobs, cs content.Store) {
|
||||
var (
|
||||
ticker = time.NewTicker(100 * time.Millisecond)
|
||||
statuses = map[string]statusInfo{}
|
||||
done bool
|
||||
)
|
||||
defer ticker.Stop()
|
||||
|
||||
pw, _, ctx := progress.FromContext(ctx)
|
||||
defer pw.Close()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-ctx.Done():
|
||||
done = true
|
||||
}
|
||||
|
||||
resolved := "resolved"
|
||||
if !ongoing.isResolved() {
|
||||
resolved = "resolving"
|
||||
}
|
||||
statuses[ongoing.name] = statusInfo{
|
||||
Ref: ongoing.name,
|
||||
Status: resolved,
|
||||
}
|
||||
|
||||
actives := make(map[string]statusInfo)
|
||||
|
||||
if !done {
|
||||
active, err := cs.ListStatuses(ctx, "")
|
||||
if err != nil {
|
||||
// log.G(ctx).WithError(err).Error("active check failed")
|
||||
continue
|
||||
}
|
||||
// update status of active entries!
|
||||
for _, active := range active {
|
||||
actives[active.Ref] = statusInfo{
|
||||
Ref: active.Ref,
|
||||
Status: "downloading",
|
||||
Offset: active.Offset,
|
||||
Total: active.Total,
|
||||
StartedAt: active.StartedAt,
|
||||
UpdatedAt: active.UpdatedAt,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// now, update the items in jobs that are not in active
|
||||
for _, j := range ongoing.jobs() {
|
||||
refKey := remotes.MakeRefKey(ctx, j.Descriptor)
|
||||
if a, ok := actives[refKey]; ok {
|
||||
started := j.started
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: a.Status,
|
||||
Total: int(a.Total),
|
||||
Current: int(a.Offset),
|
||||
Started: &started,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
if !j.done {
|
||||
info, err := cs.Info(context.TODO(), j.Digest)
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: "waiting",
|
||||
})
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
j.done = true
|
||||
}
|
||||
|
||||
if done || j.done {
|
||||
started := j.started
|
||||
createdAt := info.CreatedAt
|
||||
pw.Write(j.Digest.String(), progress.Status{
|
||||
Action: "done",
|
||||
Current: int(info.Size),
|
||||
Total: int(info.Size),
|
||||
Completed: &createdAt,
|
||||
Started: &started,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
if done {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// jobs provides a way of identifying the download keys for a particular task
|
||||
// encountering during the pull walk.
|
||||
//
|
||||
// This is very minimal and will probably be replaced with something more
|
||||
// featured.
|
||||
type jobs struct {
|
||||
name string
|
||||
added map[digest.Digest]job
|
||||
mu sync.Mutex
|
||||
resolved bool
|
||||
}
|
||||
|
||||
type job struct {
|
||||
ocispec.Descriptor
|
||||
done bool
|
||||
started time.Time
|
||||
}
|
||||
|
||||
func newJobs(name string) *jobs {
|
||||
return &jobs{
|
||||
name: name,
|
||||
added: make(map[digest.Digest]job),
|
||||
}
|
||||
}
|
||||
|
||||
func (j *jobs) add(desc ocispec.Descriptor) {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
if _, ok := j.added[desc.Digest]; ok {
|
||||
return
|
||||
}
|
||||
j.added[desc.Digest] = job{
|
||||
Descriptor: desc,
|
||||
started: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
func (j *jobs) remove(desc ocispec.Descriptor) {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
delete(j.added, desc.Digest)
|
||||
}
|
||||
|
||||
func (j *jobs) jobs() []job {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
|
||||
descs := make([]job, 0, len(j.added))
|
||||
for _, j := range j.added {
|
||||
descs = append(descs, j)
|
||||
}
|
||||
return descs
|
||||
}
|
||||
|
||||
func (j *jobs) isResolved() bool {
|
||||
j.mu.Lock()
|
||||
defer j.mu.Unlock()
|
||||
return j.resolved
|
||||
}
|
||||
|
||||
type statusInfo struct {
|
||||
Ref string
|
||||
Status string
|
||||
Offset int64
|
||||
Total int64
|
||||
StartedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
func oneOffProgress(ctx context.Context, id string) func(err error) error {
|
||||
pw, _, _ := progress.FromContext(ctx)
|
||||
now := time.Now()
|
||||
st := progress.Status{
|
||||
Started: &now,
|
||||
}
|
||||
pw.Write(id, st)
|
||||
return func(err error) error {
|
||||
// TODO: set error on status
|
||||
now := time.Now()
|
||||
st.Completed = &now
|
||||
pw.Write(id, st)
|
||||
pw.Close()
|
||||
return err
|
||||
}
|
||||
}
|
||||
75
util/pull/resolver.go
Normal file
75
util/pull/resolver.go
Normal file
@@ -0,0 +1,75 @@
|
||||
package pull
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/remotes"
|
||||
"github.com/containerd/containerd/remotes/docker"
|
||||
"github.com/moby/buildkit/session"
|
||||
"github.com/moby/buildkit/session/auth"
|
||||
"github.com/moby/buildkit/util/tracing"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
)
|
||||
|
||||
func NewResolver(ctx context.Context, sm *session.Manager, imageStore images.Store) remotes.Resolver {
|
||||
r := docker.NewResolver(docker.ResolverOptions{
|
||||
Client: tracing.DefaultClient,
|
||||
Credentials: getCredentialsFromSession(ctx, sm),
|
||||
})
|
||||
|
||||
if imageStore == nil {
|
||||
return r
|
||||
}
|
||||
|
||||
return localFallbackResolver{r, imageStore}
|
||||
}
|
||||
|
||||
func getCredentialsFromSession(ctx context.Context, sm *session.Manager) func(string) (string, string, error) {
|
||||
id := session.FromContext(ctx)
|
||||
if id == "" {
|
||||
return nil
|
||||
}
|
||||
return func(host string) (string, string, error) {
|
||||
timeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
caller, err := sm.Get(timeoutCtx, id)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
return auth.CredentialsFunc(tracing.ContextWithSpanFromContext(context.TODO(), ctx), caller)(host)
|
||||
}
|
||||
}
|
||||
|
||||
// A remotes.Resolver which checks the local image store if the real
|
||||
// resolver cannot find the image, essentially falling back to a local
|
||||
// image if one is present.
|
||||
//
|
||||
// We do not override the Fetcher or Pusher methods:
|
||||
//
|
||||
// - Fetcher is called by github.com/containerd/containerd/remotes/:fetch()
|
||||
// only after it has checked for the content locally, so avoid the
|
||||
// hassle of interposing a local-fetch proxy and simply pass on the
|
||||
// request.
|
||||
// - Pusher wouldn't make sense to push locally, so just forward.
|
||||
|
||||
type localFallbackResolver struct {
|
||||
remotes.Resolver
|
||||
is images.Store
|
||||
}
|
||||
|
||||
func (r localFallbackResolver) Resolve(ctx context.Context, ref string) (string, ocispec.Descriptor, error) {
|
||||
n, desc, err := r.Resolver.Resolve(ctx, ref)
|
||||
if err == nil {
|
||||
return n, desc, err
|
||||
}
|
||||
|
||||
img, err2 := r.is.Get(ctx, ref)
|
||||
if err2 != nil {
|
||||
return "", ocispec.Descriptor{}, err
|
||||
}
|
||||
return ref, img.Target, nil
|
||||
}
|
||||
Reference in New Issue
Block a user