push_v1.go

package distribution

import (
	"fmt"
	"sync"

	"github.com/Sirupsen/logrus"
	"github.com/docker/distribution/digest"
	"github.com/docker/distribution/registry/client/transport"
	"github.com/hyperhq/hypercli/distribution/metadata"
	"github.com/hyperhq/hypercli/dockerversion"
	"github.com/hyperhq/hypercli/image"
	"github.com/hyperhq/hypercli/image/v1"
	"github.com/hyperhq/hypercli/layer"
	"github.com/hyperhq/hypercli/pkg/ioutils"
	"github.com/hyperhq/hypercli/pkg/progress"
	"github.com/hyperhq/hypercli/pkg/stringid"
	"github.com/hyperhq/hypercli/reference"
	"github.com/hyperhq/hypercli/registry"
	"golang.org/x/net/context"
)

type v1Pusher struct {
	ctx         context.Context
	v1IDService *metadata.V1IDService
	endpoint    registry.APIEndpoint
	ref         reference.Named
	repoInfo    *registry.RepositoryInfo
	config      *ImagePushConfig
	session     *registry.Session
}

func (p *v1Pusher) Push(ctx context.Context) error {
	tlsConfig, err := p.config.RegistryService.TLSConfig(p.repoInfo.Index.Name)
	if err != nil {
		return err
	}
	// Adds Docker-specific headers as well as user-specified headers (metaHeaders)
	tr := transport.NewTransport(
		// TODO(tiborvass): was NoTimeout
		registry.NewTransport(tlsConfig),
		registry.DockerHeaders(dockerversion.DockerUserAgent(), p.config.MetaHeaders)...,
	)
	client := registry.HTTPClient(tr)
	v1Endpoint, err := p.endpoint.ToV1Endpoint(dockerversion.DockerUserAgent(), p.config.MetaHeaders)
	if err != nil {
		logrus.Debugf("Could not get v1 endpoint: %v", err)
		return fallbackError{err: err}
	}
	p.session, err = registry.NewSession(client, p.config.AuthConfig, v1Endpoint)
	if err != nil {
		// TODO(dmcgowan): Check if should fallback
		return fallbackError{err: err}
	}
	if err := p.pushRepository(ctx); err != nil {
		// TODO(dmcgowan): Check if should fallback
		return err
	}
	return nil
}

// v1Image exposes the configuration, filesystem layer ID, and a v1 ID for an
// image being pushed to a v1 registry.
type v1Image interface {
	Config() []byte
	Layer() layer.Layer
	V1ID() string
}

type v1ImageCommon struct {
	layer  layer.Layer
	config []byte
	v1ID   string
}

func (common *v1ImageCommon) Config() []byte {
	return common.config
}

func (common *v1ImageCommon) V1ID() string {
	return common.v1ID
}

func (common *v1ImageCommon) Layer() layer.Layer {
	return common.layer
}

// v1TopImage defines a runnable (top layer) image being pushed to a v1
// registry.
type v1TopImage struct {
	v1ImageCommon
	imageID image.ID
}

func newV1TopImage(imageID image.ID, img *image.Image, l layer.Layer, parent *v1DependencyImage) (*v1TopImage, error) {
	v1ID := digest.Digest(imageID).Hex()
	parentV1ID := ""
	if parent != nil {
		parentV1ID = parent.V1ID()
	}

	config, err := v1.MakeV1ConfigFromConfig(img, v1ID, parentV1ID, false)
	if err != nil {
		return nil, err
	}

	return &v1TopImage{
		v1ImageCommon: v1ImageCommon{
			v1ID:   v1ID,
			config: config,
			layer:  l,
		},
		imageID: imageID,
	}, nil
}

// v1DependencyImage defines a dependency layer being pushed to a v1 registry.
type v1DependencyImage struct {
	v1ImageCommon
}

func newV1DependencyImage(l layer.Layer, parent *v1DependencyImage) (*v1DependencyImage, error) {
	v1ID := digest.Digest(l.ChainID()).Hex()

	config := ""
	if parent != nil {
		config = fmt.Sprintf(`{"id":"%s","parent":"%s"}`, v1ID, parent.V1ID())
	} else {
		config = fmt.Sprintf(`{"id":"%s"}`, v1ID)
	}
	return &v1DependencyImage{
		v1ImageCommon: v1ImageCommon{
			v1ID:   v1ID,
			config: []byte(config),
			layer:  l,
		},
	}, nil
}

// Retrieve the all the images to be uploaded in the correct order
func (p *v1Pusher) getImageList() (imageList []v1Image, tagsByImage map[image.ID][]string, referencedLayers []layer.Layer, err error) {
	tagsByImage = make(map[image.ID][]string)

	// Ignore digest references
	if _, isCanonical := p.ref.(reference.Canonical); isCanonical {
		return
	}

	tagged, isTagged := p.ref.(reference.NamedTagged)
	if isTagged {
		// Push a specific tag
		var imgID image.ID
		imgID, err = p.config.ReferenceStore.Get(p.ref)
		if err != nil {
			return
		}

		imageList, err = p.imageListForTag(imgID, nil, &referencedLayers)
		if err != nil {
			return
		}

		tagsByImage[imgID] = []string{tagged.Tag()}

		return
	}

	imagesSeen := make(map[image.ID]struct{})
	dependenciesSeen := make(map[layer.ChainID]*v1DependencyImage)

	associations := p.config.ReferenceStore.ReferencesByName(p.ref)
	for _, association := range associations {
		if tagged, isTagged = association.Ref.(reference.NamedTagged); !isTagged {
			// Ignore digest references.
			continue
		}

		tagsByImage[association.ImageID] = append(tagsByImage[association.ImageID], tagged.Tag())

		if _, present := imagesSeen[association.ImageID]; present {
			// Skip generating image list for already-seen image
			continue
		}
		imagesSeen[association.ImageID] = struct{}{}

		imageListForThisTag, err := p.imageListForTag(association.ImageID, dependenciesSeen, &referencedLayers)
		if err != nil {
			return nil, nil, nil, err
		}

		// append to main image list
		imageList = append(imageList, imageListForThisTag...)
	}
	if len(imageList) == 0 {
		return nil, nil, nil, fmt.Errorf("No images found for the requested repository / tag")
	}
	logrus.Debugf("Image list: %v", imageList)
	logrus.Debugf("Tags by image: %v", tagsByImage)

	return
}

func (p *v1Pusher) imageListForTag(imgID image.ID, dependenciesSeen map[layer.ChainID]*v1DependencyImage, referencedLayers *[]layer.Layer) (imageListForThisTag []v1Image, err error) {
	img, err := p.config.ImageStore.Get(imgID)
	if err != nil {
		return nil, err
	}

	topLayerID := img.RootFS.ChainID()

	var l layer.Layer
	if topLayerID == "" {
		l = layer.EmptyLayer
	} else {
		l, err = p.config.LayerStore.Get(topLayerID)
		*referencedLayers = append(*referencedLayers, l)
		if err != nil {
			return nil, fmt.Errorf("failed to get top layer from image: %v", err)
		}
	}

	dependencyImages, parent, err := generateDependencyImages(l.Parent(), dependenciesSeen)
	if err != nil {
		return nil, err
	}

	topImage, err := newV1TopImage(imgID, img, l, parent)
	if err != nil {
		return nil, err
	}

	imageListForThisTag = append(dependencyImages, topImage)

	return
}

func generateDependencyImages(l layer.Layer, dependenciesSeen map[layer.ChainID]*v1DependencyImage) (imageListForThisTag []v1Image, parent *v1DependencyImage, err error) {
	if l == nil {
		return nil, nil, nil
	}

	imageListForThisTag, parent, err = generateDependencyImages(l.Parent(), dependenciesSeen)

	if dependenciesSeen != nil {
		if dependencyImage, present := dependenciesSeen[l.ChainID()]; present {
			// This layer is already on the list, we can ignore it
			// and all its parents.
			return imageListForThisTag, dependencyImage, nil
		}
	}

	dependencyImage, err := newV1DependencyImage(l, parent)
	if err != nil {
		return nil, nil, err
	}
	imageListForThisTag = append(imageListForThisTag, dependencyImage)

	if dependenciesSeen != nil {
		dependenciesSeen[l.ChainID()] = dependencyImage
	}

	return imageListForThisTag, dependencyImage, nil
}

// createImageIndex returns an index of an image's layer IDs and tags.
func createImageIndex(images []v1Image, tags map[image.ID][]string) []*registry.ImgData {
	var imageIndex []*registry.ImgData
	for _, img := range images {
		v1ID := img.V1ID()

		if topImage, isTopImage := img.(*v1TopImage); isTopImage {
			if tags, hasTags := tags[topImage.imageID]; hasTags {
				// If an image has tags you must add an entry in the image index
				// for each tag
				for _, tag := range tags {
					imageIndex = append(imageIndex, &registry.ImgData{
						ID:  v1ID,
						Tag: tag,
					})
				}
				continue
			}
		}

		// If the image does not have a tag it still needs to be sent to the
		// registry with an empty tag so that it is associated with the repository
		imageIndex = append(imageIndex, &registry.ImgData{
			ID:  v1ID,
			Tag: "",
		})
	}
	return imageIndex
}

// lookupImageOnEndpoint checks the specified endpoint to see if an image exists
// and if it is absent then it sends the image id to the channel to be pushed.
func (p *v1Pusher) lookupImageOnEndpoint(wg *sync.WaitGroup, endpoint string, images chan v1Image, imagesToPush chan string) {
	defer wg.Done()
	for image := range images {
		v1ID := image.V1ID()
		truncID := stringid.TruncateID(image.Layer().DiffID().String())
		if err := p.session.LookupRemoteImage(v1ID, endpoint); err != nil {
			logrus.Errorf("Error in LookupRemoteImage: %s", err)
			imagesToPush <- v1ID
			progress.Update(p.config.ProgressOutput, truncID, "Waiting")
		} else {
			progress.Update(p.config.ProgressOutput, truncID, "Already exists")
		}
	}
}

func (p *v1Pusher) pushImageToEndpoint(ctx context.Context, endpoint string, imageList []v1Image, tags map[image.ID][]string, repo *registry.RepositoryData) error {
	workerCount := len(imageList)
	// start a maximum of 5 workers to check if images exist on the specified endpoint.
	if workerCount > 5 {
		workerCount = 5
	}
	var (
		wg           = &sync.WaitGroup{}
		imageData    = make(chan v1Image, workerCount*2)
		imagesToPush = make(chan string, workerCount*2)
		pushes       = make(chan map[string]struct{}, 1)
	)
	for i := 0; i < workerCount; i++ {
		wg.Add(1)
		go p.lookupImageOnEndpoint(wg, endpoint, imageData, imagesToPush)
	}
	// start a go routine that consumes the images to push
	go func() {
		shouldPush := make(map[string]struct{})
		for id := range imagesToPush {
			shouldPush[id] = struct{}{}
		}
		pushes <- shouldPush
	}()
	for _, v1Image := range imageList {
		imageData <- v1Image
	}
	// close the channel to notify the workers that there will be no more images to check.
	close(imageData)
	wg.Wait()
	close(imagesToPush)
	// wait for all the images that require pushes to be collected into a consumable map.
	shouldPush := <-pushes
	// finish by pushing any images and tags to the endpoint.  The order that the images are pushed
	// is very important that is why we are still iterating over the ordered list of imageIDs.
	for _, img := range imageList {
		v1ID := img.V1ID()
		if _, push := shouldPush[v1ID]; push {
			if _, err := p.pushImage(ctx, img, endpoint); err != nil {
				// FIXME: Continue on error?
				return err
			}
		}
		if topImage, isTopImage := img.(*v1TopImage); isTopImage {
			for _, tag := range tags[topImage.imageID] {
				progress.Messagef(p.config.ProgressOutput, "", "Pushing tag for rev [%s] on {%s}", stringid.TruncateID(v1ID), endpoint+"repositories/"+p.repoInfo.RemoteName()+"/tags/"+tag)
				if err := p.session.PushRegistryTag(p.repoInfo, v1ID, tag, endpoint); err != nil {
					return err
				}
			}
		}
	}
	return nil
}

// pushRepository pushes layers that do not already exist on the registry.
func (p *v1Pusher) pushRepository(ctx context.Context) error {
	imgList, tags, referencedLayers, err := p.getImageList()
	defer func() {
		for _, l := range referencedLayers {
			p.config.LayerStore.Release(l)
		}
	}()
	if err != nil {
		return err
	}

	imageIndex := createImageIndex(imgList, tags)
	for _, data := range imageIndex {
		logrus.Debugf("Pushing ID: %s with Tag: %s", data.ID, data.Tag)
	}

	// Register all the images in a repository with the registry
	// If an image is not in this list it will not be associated with the repository
	repoData, err := p.session.PushImageJSONIndex(p.repoInfo, imageIndex, false, nil)
	if err != nil {
		return err
	}
	// push the repository to each of the endpoints only if it does not exist.
	for _, endpoint := range repoData.Endpoints {
		if err := p.pushImageToEndpoint(ctx, endpoint, imgList, tags, repoData); err != nil {
			return err
		}
	}
	_, err = p.session.PushImageJSONIndex(p.repoInfo, imageIndex, true, repoData.Endpoints)
	return err
}

func (p *v1Pusher) pushImage(ctx context.Context, v1Image v1Image, ep string) (checksum string, err error) {
	l := v1Image.Layer()
	v1ID := v1Image.V1ID()
	truncID := stringid.TruncateID(l.DiffID().String())

	jsonRaw := v1Image.Config()
	progress.Update(p.config.ProgressOutput, truncID, "Pushing")

	// General rule is to use ID for graph accesses and compatibilityID for
	// calls to session.registry()
	imgData := &registry.ImgData{
		ID: v1ID,
	}

	// Send the json
	if err := p.session.PushImageJSONRegistry(imgData, jsonRaw, ep); err != nil {
		if err == registry.ErrAlreadyExists {
			progress.Update(p.config.ProgressOutput, truncID, "Image already pushed, skipping")
			return "", nil
		}
		return "", err
	}

	arch, err := l.TarStream()
	if err != nil {
		return "", err
	}
	defer arch.Close()

	// don't care if this fails; best effort
	size, _ := l.DiffSize()

	// Send the layer
	logrus.Debugf("rendered layer for %s of [%d] size", v1ID, size)

	reader := progress.NewProgressReader(ioutils.NewCancelReadCloser(ctx, arch), p.config.ProgressOutput, size, truncID, "Pushing")
	defer reader.Close()

	checksum, checksumPayload, err := p.session.PushImageLayerRegistry(v1ID, reader, ep, jsonRaw)
	if err != nil {
		return "", err
	}
	imgData.Checksum = checksum
	imgData.ChecksumPayload = checksumPayload
	// Send the checksum
	if err := p.session.PushImageChecksumRegistry(imgData, ep); err != nil {
		return "", err
	}

	if err := p.v1IDService.Set(v1ID, p.repoInfo.Index.Name, l.DiffID()); err != nil {
		logrus.Warnf("Could not set v1 ID mapping: %v", err)
	}

	progress.Update(p.config.ProgressOutput, truncID, "Image successfully pushed")
	return imgData.Checksum, nil
}