Method PutBlobWithOptions

docker/docker_image_dest.go:138–228

PutBlobWithOptions writes contents of stream and returns data representing the result. inputInfo.Digest can be optionally provided if known; if provided, and stream is read to the end without error, the digest MUST match the stream contents. inputInfo.Size is the expected length of stream, if known.

(ctx context.Context, stream io.Reader, inputInfo types.BlobInfo, options private.PutBlobOptions)

// to any other readers for download using the supplied digest.
// If stream.Read() at any time, ESPECIALLY at end of input, returns an error, PutBlobWithOptions MUST 1) fail, and 2) delete any data stored so far.
func (d *dockerImageDestination) PutBlobWithOptions(ctx context.Context, stream io.Reader, inputInfo types.BlobInfo, options private.PutBlobOptions) (private.UploadedBlob, error) {
	// If requested, precompute the blob digest to prevent uploading layers that already exist on the registry.
	// This functionality is particularly useful when BlobInfoCache has not been populated with compressed digests,
	// the source blob is uncompressed, and the destination blob is being compressed "on the fly".
	if inputInfo.Digest == "" && d.c.sys != nil && d.c.sys.DockerRegistryPushPrecomputeDigests {
		logrus.Debugf("Precomputing digest layer for %s", reference.Path(d.ref.ref))
		streamCopy, cleanup, err := streamdigest.ComputeBlobInfo(d.c.sys, stream, &inputInfo)
		if err != nil {
			return private.UploadedBlob{}, err
		}
		defer cleanup()
		stream = streamCopy
	}

	if inputInfo.Digest != "" {
		// This should not really be necessary, at least the copy code calls TryReusingBlob automatically.
		// Still, we need to check, if only because the "initiate upload" endpoint does not have a documented "blob already exists" return value.
		haveBlob, reusedInfo, err := d.tryReusingExactBlob(ctx, inputInfo, options.Cache)
		if err != nil {
			return private.UploadedBlob{}, err
		}
		if haveBlob {
			return private.UploadedBlob{Digest: reusedInfo.Digest, Size: reusedInfo.Size}, nil
		}
	}

	// FIXME? Chunked upload, progress reporting, etc.
	uploadPath := fmt.Sprintf(blobUploadPath, reference.Path(d.ref.ref))
	logrus.Debugf("Uploading %s", uploadPath)
	res, err := d.c.makeRequest(ctx, http.MethodPost, uploadPath, nil, nil, v2Auth, nil)
	if err != nil {
		return private.UploadedBlob{}, err
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusAccepted {
		logrus.Debugf("Error initiating layer upload, response %#v", *res)
		return private.UploadedBlob{}, fmt.Errorf("initiating layer upload to %s in %s: %w", uploadPath, d.c.registry, registryHTTPResponseToError(res))
	}
	uploadLocation, err := res.Location()
	if err != nil {
		return private.UploadedBlob{}, fmt.Errorf("determining upload URL: %w", err)
	}

	digester, stream := putblobdigest.DigestIfCanonicalUnknown(stream, inputInfo)
	sizeCounter := &sizeCounter{}
	stream = io.TeeReader(stream, sizeCounter)

	uploadLocation, err = func() (*url.URL, error) { // A scope for defer
		uploadReader := uploadreader.NewUploadReader(stream)
		// This error text should never be user-visible, we terminate only after makeRequestToResolvedURL
		// returns, so there isn’t a way for the error text to be provided to any of our callers.
		defer uploadReader.Terminate(errors.New("Reading data from an already terminated upload"))
		res, err = d.c.makeRequestToResolvedURL(ctx, http.MethodPatch, uploadLocation, map[string][]string{"Content-Type": {"application/octet-stream"}}, uploadReader, inputInfo.Size, v2Auth, nil)
		if err != nil {
			logrus.Debugf("Error uploading layer chunked %v", err)
			return nil, err
		}
		defer res.Body.Close()