package distribution import ( "errors" "fmt" "io" "sync" "github.com/Sirupsen/logrus" "github.com/docker/distribution" "github.com/docker/distribution/digest" "github.com/docker/distribution/manifest/schema1" "github.com/docker/distribution/manifest/schema2" distreference "github.com/docker/distribution/reference" "github.com/docker/distribution/registry/client" "github.com/docker/docker/distribution/metadata" "github.com/docker/docker/distribution/xfer" "github.com/docker/docker/image" "github.com/docker/docker/layer" "github.com/docker/docker/pkg/ioutils" "github.com/docker/docker/pkg/progress" "github.com/docker/docker/pkg/stringid" "github.com/docker/docker/reference" "github.com/docker/docker/registry" "golang.org/x/net/context" ) // PushResult contains the tag, manifest digest, and manifest size from the // push. It's used to signal this information to the trust code in the client // so it can sign the manifest if necessary. type PushResult struct { Tag string Digest digest.Digest Size int } type v2Pusher struct { v2MetadataService *metadata.V2MetadataService ref reference.Named endpoint registry.APIEndpoint repoInfo *registry.RepositoryInfo config *ImagePushConfig repo distribution.Repository // pushState is state built by the Upload functions. pushState pushState } type pushState struct { sync.Mutex // remoteLayers is the set of layers known to exist on the remote side. // This avoids redundant queries when pushing multiple tags that // involve the same layers. It is also used to fill in digest and size // information when building the manifest. remoteLayers map[layer.DiffID]distribution.Descriptor // confirmedV2 is set to true if we confirm we're talking to a v2 // registry. This is used to limit fallbacks to the v1 protocol. confirmedV2 bool } func (p *v2Pusher) Push(ctx context.Context) (err error) { p.pushState.remoteLayers = make(map[layer.DiffID]distribution.Descriptor) p.repo, p.pushState.confirmedV2, err = NewV2Repository(ctx, p.repoInfo, p.endpoint, p.config.MetaHeaders, p.config.AuthConfig, "push", "pull") if err != nil { logrus.Debugf("Error getting v2 registry: %v", err) return err } if err = p.pushV2Repository(ctx); err != nil { if continueOnError(err) { return fallbackError{ err: err, confirmedV2: p.pushState.confirmedV2, transportOK: true, } } } return err } func (p *v2Pusher) pushV2Repository(ctx context.Context) (err error) { if namedTagged, isNamedTagged := p.ref.(reference.NamedTagged); isNamedTagged { imageID, err := p.config.ReferenceStore.Get(p.ref) if err != nil { return fmt.Errorf("tag does not exist: %s", p.ref.String()) } return p.pushV2Tag(ctx, namedTagged, imageID) } if !reference.IsNameOnly(p.ref) { return errors.New("cannot push a digest reference") } // Pull all tags pushed := 0 for _, association := range p.config.ReferenceStore.ReferencesByName(p.ref) { if namedTagged, isNamedTagged := association.Ref.(reference.NamedTagged); isNamedTagged { pushed++ if err := p.pushV2Tag(ctx, namedTagged, association.ImageID); err != nil { return err } } } if pushed == 0 { return fmt.Errorf("no tags to push for %s", p.repoInfo.Name()) } return nil } func (p *v2Pusher) pushV2Tag(ctx context.Context, ref reference.NamedTagged, imageID image.ID) error { logrus.Debugf("Pushing repository: %s", ref.String()) img, err := p.config.ImageStore.Get(imageID) if err != nil { return fmt.Errorf("could not find image from tag %s: %v", ref.String(), err) } var l layer.Layer topLayerID := img.RootFS.ChainID() if topLayerID == "" { l = layer.EmptyLayer } else { l, err = p.config.LayerStore.Get(topLayerID) if err != nil { return fmt.Errorf("failed to get top layer from image: %v", err) } defer layer.ReleaseAndLog(p.config.LayerStore, l) } var descriptors []xfer.UploadDescriptor descriptorTemplate := v2PushDescriptor{ v2MetadataService: p.v2MetadataService, repoInfo: p.repoInfo, repo: p.repo, pushState: &p.pushState, } // Loop bounds condition is to avoid pushing the base layer on Windows. for i := 0; i < len(img.RootFS.DiffIDs); i++ { descriptor := descriptorTemplate descriptor.layer = l descriptors = append(descriptors, &descriptor) l = l.Parent() } if err := p.config.UploadManager.Upload(ctx, descriptors, p.config.ProgressOutput); err != nil { return err } // Try schema2 first builder := schema2.NewManifestBuilder(p.repo.Blobs(ctx), img.RawJSON()) manifest, err := manifestFromBuilder(ctx, builder, descriptors) if err != nil { return err } manSvc, err := p.repo.Manifests(ctx) if err != nil { return err } putOptions := []distribution.ManifestServiceOption{client.WithTag(ref.Tag())} if _, err = manSvc.Put(ctx, manifest, putOptions...); err != nil { logrus.Warnf("failed to upload schema2 manifest: %v - falling back to schema1", err) manifestRef, err := distreference.WithTag(p.repo.Named(), ref.Tag()) if err != nil { return err } builder = schema1.NewConfigManifestBuilder(p.repo.Blobs(ctx), p.config.TrustKey, manifestRef, img.RawJSON()) manifest, err = manifestFromBuilder(ctx, builder, descriptors) if err != nil { return err } if _, err = manSvc.Put(ctx, manifest, putOptions...); err != nil { return err } } var canonicalManifest []byte switch v := manifest.(type) { case *schema1.SignedManifest: canonicalManifest = v.Canonical case *schema2.DeserializedManifest: _, canonicalManifest, err = v.Payload() if err != nil { return err } } manifestDigest := digest.FromBytes(canonicalManifest) progress.Messagef(p.config.ProgressOutput, "", "%s: digest: %s size: %d", ref.Tag(), manifestDigest, len(canonicalManifest)) // Signal digest to the trust client so it can sign the // push, if appropriate. progress.Aux(p.config.ProgressOutput, PushResult{Tag: ref.Tag(), Digest: manifestDigest, Size: len(canonicalManifest)}) return nil } func manifestFromBuilder(ctx context.Context, builder distribution.ManifestBuilder, descriptors []xfer.UploadDescriptor) (distribution.Manifest, error) { // descriptors is in reverse order; iterate backwards to get references // appended in the right order. for i := len(descriptors) - 1; i >= 0; i-- { if err := builder.AppendReference(descriptors[i].(*v2PushDescriptor)); err != nil { return nil, err } } return builder.Build(ctx) } type v2PushDescriptor struct { layer layer.Layer v2MetadataService *metadata.V2MetadataService repoInfo reference.Named repo distribution.Repository pushState *pushState remoteDescriptor distribution.Descriptor } func (pd *v2PushDescriptor) Key() string { return "v2push:" + pd.repo.Named().Name() + " " + pd.layer.DiffID().String() } func (pd *v2PushDescriptor) ID() string { return stringid.TruncateID(pd.layer.DiffID().String()) } func (pd *v2PushDescriptor) DiffID() layer.DiffID { return pd.layer.DiffID() } func (pd *v2PushDescriptor) Upload(ctx context.Context, progressOutput progress.Output) (distribution.Descriptor, error) { diffID := pd.DiffID() pd.pushState.Lock() if descriptor, ok := pd.pushState.remoteLayers[diffID]; ok { // it is already known that the push is not needed and // therefore doing a stat is unnecessary pd.pushState.Unlock() progress.Update(progressOutput, pd.ID(), "Layer already exists") return descriptor, nil } pd.pushState.Unlock() // Do we have any metadata associated with this layer's DiffID? v2Metadata, err := pd.v2MetadataService.GetMetadata(diffID) if err == nil { descriptor, exists, err := layerAlreadyExists(ctx, v2Metadata, pd.repoInfo, pd.repo, pd.pushState) if err != nil { progress.Update(progressOutput, pd.ID(), "Image push failed") return distribution.Descriptor{}, retryOnError(err) } if exists { progress.Update(progressOutput, pd.ID(), "Layer already exists") pd.pushState.Lock() pd.pushState.remoteLayers[diffID] = descriptor pd.pushState.Unlock() return descriptor, nil } } logrus.Debugf("Pushing layer: %s", diffID) // if digest was empty or not saved, or if blob does not exist on the remote repository, // then push the blob. bs := pd.repo.Blobs(ctx) var layerUpload distribution.BlobWriter mountAttemptsRemaining := 3 // Attempt to find another repository in the same registry to mount the layer // from to avoid an unnecessary upload. // Note: metadata is stored from oldest to newest, so we iterate through this // slice in reverse to maximize our chances of the blob still existing in the // remote repository. for i := len(v2Metadata) - 1; i >= 0 && mountAttemptsRemaining > 0; i-- { mountFrom := v2Metadata[i] sourceRepo, err := reference.ParseNamed(mountFrom.SourceRepository) if err != nil { continue } if pd.repoInfo.Hostname() != sourceRepo.Hostname() { // don't mount blobs from another registry continue } namedRef, err := reference.WithName(mountFrom.SourceRepository) if err != nil { continue } // TODO (brianbland): We need to construct a reference where the Name is // only the full remote name, so clean this up when distribution has a // richer reference package remoteRef, err := distreference.WithName(namedRef.RemoteName()) if err != nil { continue } canonicalRef, err := distreference.WithDigest(remoteRef, mountFrom.Digest) if err != nil { continue } logrus.Debugf("attempting to mount layer %s (%s) from %s", diffID, mountFrom.Digest, sourceRepo.FullName()) layerUpload, err = bs.Create(ctx, client.WithMountFrom(canonicalRef)) switch err := err.(type) { case distribution.ErrBlobMounted: progress.Updatef(progressOutput, pd.ID(), "Mounted from %s", err.From.Name()) err.Descriptor.MediaType = schema2.MediaTypeLayer pd.pushState.Lock() pd.pushState.confirmedV2 = true pd.pushState.remoteLayers[diffID] = err.Descriptor pd.pushState.Unlock() // Cache mapping from this layer's DiffID to the blobsum if err := pd.v2MetadataService.Add(diffID, metadata.V2Metadata{Digest: mountFrom.Digest, SourceRepository: pd.repoInfo.FullName()}); err != nil { return distribution.Descriptor{}, xfer.DoNotRetry{Err: err} } return err.Descriptor, nil case nil: // blob upload session created successfully, so begin the upload mountAttemptsRemaining = 0 default: // unable to mount layer from this repository, so this source mapping is no longer valid logrus.Debugf("unassociating layer %s (%s) with %s", diffID, mountFrom.Digest, mountFrom.SourceRepository) pd.v2MetadataService.Remove(mountFrom) mountAttemptsRemaining-- } } if layerUpload == nil { layerUpload, err = bs.Create(ctx) if err != nil { return distribution.Descriptor{}, retryOnError(err) } } defer layerUpload.Close() arch, err := pd.layer.TarStream() if err != nil { return distribution.Descriptor{}, xfer.DoNotRetry{Err: err} } // don't care if this fails; best effort size, _ := pd.layer.DiffSize() reader := progress.NewProgressReader(ioutils.NewCancelReadCloser(ctx, arch), progressOutput, size, pd.ID(), "Pushing") compressedReader, compressionDone := compress(reader) defer func() { reader.Close() <-compressionDone }() digester := digest.Canonical.New() tee := io.TeeReader(compressedReader, digester.Hash()) nn, err := layerUpload.ReadFrom(tee) compressedReader.Close() if err != nil { return distribution.Descriptor{}, retryOnError(err) } pushDigest := digester.Digest() if _, err := layerUpload.Commit(ctx, distribution.Descriptor{Digest: pushDigest}); err != nil { return distribution.Descriptor{}, retryOnError(err) } logrus.Debugf("uploaded layer %s (%s), %d bytes", diffID, pushDigest, nn) progress.Update(progressOutput, pd.ID(), "Pushed") // Cache mapping from this layer's DiffID to the blobsum if err := pd.v2MetadataService.Add(diffID, metadata.V2Metadata{Digest: pushDigest, SourceRepository: pd.repoInfo.FullName()}); err != nil { return distribution.Descriptor{}, xfer.DoNotRetry{Err: err} } pd.pushState.Lock() // If Commit succeeded, that's an indication that the remote registry // speaks the v2 protocol. pd.pushState.confirmedV2 = true descriptor := distribution.Descriptor{ Digest: pushDigest, MediaType: schema2.MediaTypeLayer, Size: nn, } pd.pushState.remoteLayers[diffID] = descriptor pd.pushState.Unlock() return descriptor, nil } func (pd *v2PushDescriptor) SetRemoteDescriptor(descriptor distribution.Descriptor) { pd.remoteDescriptor = descriptor } func (pd *v2PushDescriptor) Descriptor() distribution.Descriptor { return pd.remoteDescriptor } // layerAlreadyExists checks if the registry already know about any of the // metadata passed in the "metadata" slice. If it finds one that the registry // knows about, it returns the known digest and "true". func layerAlreadyExists(ctx context.Context, metadata []metadata.V2Metadata, repoInfo reference.Named, repo distribution.Repository, pushState *pushState) (distribution.Descriptor, bool, error) { for _, meta := range metadata { // Only check blobsums that are known to this repository or have an unknown source if meta.SourceRepository != "" && meta.SourceRepository != repoInfo.FullName() { continue } descriptor, err := repo.Blobs(ctx).Stat(ctx, meta.Digest) switch err { case nil: descriptor.MediaType = schema2.MediaTypeLayer return descriptor, true, nil case distribution.ErrBlobUnknown: // nop default: return distribution.Descriptor{}, false, err } } return distribution.Descriptor{}, false, nil }