vbatts/registry
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Updates client to newer routes and changes "layer" to "blob"

Browse source
This commit is contained in:
Brian Bland 2014-11-19 18:06:54 -08:00
parent b65d8d046e
commit 1336ced030
6 changed files with 238 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

153
client/client.go
View file

@ -16,57 +16,59 @@ import (
// Client implements the client interface to the registry http api
type Client interface {
// GetImageManifest returns an image manifest for the image at the given
// name, tag pair
// name, tag pair.
GetImageManifest(name, tag string) (*registry.ImageManifest, error)
// PutImageManifest uploads an image manifest for the image at the given
// name, tag pair
// name, tag pair.
PutImageManifest(name, tag string, imageManifest *registry.ImageManifest) error
// DeleteImage removes the image at the given name, tag pair
// DeleteImage removes the image at the given name, tag pair.
DeleteImage(name, tag string) error
// ListImageTags returns a list of all image tags with the given repository
// name
// name.
ListImageTags(name string) ([]string, error)
// GetImageLayer returns the image layer at the given name, tarsum pair in
// the form of an io.ReadCloser with the length of this layer
// A nonzero byteOffset can be provided to receive a partial layer beginning
// at the given offset
GetImageLayer(name, tarsum string, byteOffset int) (io.ReadCloser, int, error)
// BlobLength returns the length of the blob stored at the given name,
// digest pair.
// Returns a length value of -1 on error or if the blob does not exist.
BlobLength(name, digest string) (int, error)
// InitiateLayerUpload starts an image upload for the given name, tarsum
// pair and returns a unique location url to use for other layer upload
// methods
// Returns a *registry.LayerAlreadyExistsError if the layer already exists
// on the registry
InitiateLayerUpload(name, tarsum string) (string, error)
// GetBlob returns the blob stored at the given name, digest pair in the
// form of an io.ReadCloser with the length of this blob.
// A nonzero byteOffset can be provided to receive a partial blob beginning
// at the given offset.
GetBlob(name, digest string, byteOffset int) (io.ReadCloser, int, error)
// GetLayerUploadStatus returns the byte offset and length of the layer at
// the given upload location
GetLayerUploadStatus(location string) (int, int, error)
// InitiateBlobUpload starts a blob upload in the given repository namespace
// and returns a unique location url to use for other blob upload methods.
InitiateBlobUpload(name string) (string, error)
// UploadLayer uploads a full image layer to the registry
UploadLayer(location string, layer io.ReadCloser, length int, checksum *registry.Checksum) error
// GetBlobUploadStatus returns the byte offset and length of the blob at the
// given upload location.
GetBlobUploadStatus(location string) (int, int, error)
// UploadLayerChunk uploads a layer chunk with a given length and startByte
// to the registry
// FinishChunkedLayerUpload must be called to finalize this upload
UploadLayerChunk(location string, layerChunk io.ReadCloser, length, startByte int) error
// UploadBlob uploads a full blob to the registry.
UploadBlob(location string, blob io.ReadCloser, length int, digest string) error
// FinishChunkedLayerUpload completes a chunked layer upload at a given
// location
FinishChunkedLayerUpload(location string, length int, checksum *registry.Checksum) error
// UploadBlobChunk uploads a blob chunk with a given length and startByte to
// the registry.
// FinishChunkedBlobUpload must be called to finalize this upload.
UploadBlobChunk(location string, blobChunk io.ReadCloser, length, startByte int) error
// CancelLayerUpload deletes all content at the unfinished layer upload
// location and invalidates any future calls to this layer upload
CancelLayerUpload(location string) error
// FinishChunkedBlobUpload completes a chunked blob upload at a given
// location.
FinishChunkedBlobUpload(location string, length int, digest string) error
// CancelBlobUpload deletes all content at the unfinished blob upload
// location and invalidates any future calls to this blob upload.
CancelBlobUpload(location string) error
}
// New returns a new Client which operates against a registry with the
// given base endpoint
// This endpoint should not include /v2/ or any part of the url after this
// This endpoint should not include /v2/ or any part of the url after this.
func New(endpoint string) Client {
return &clientImpl{endpoint}
}
@ -220,9 +222,41 @@ func (r *clientImpl) ListImageTags(name string) ([]string, error) {
return tags.Tags, nil
}
func (r *clientImpl) GetImageLayer(name, tarsum string, byteOffset int) (io.ReadCloser, int, error) {
func (r *clientImpl) BlobLength(name, digest string) (int, error) {
response, err := http.Head(fmt.Sprintf("%s/v2/%s/blob/%s", r.Endpoint, name, digest))
if err != nil {
return -1, err
}
defer response.Body.Close()
// TODO(bbland): handle other status codes, like 5xx errors
switch {
case response.StatusCode == http.StatusOK:
lengthHeader := response.Header.Get("Content-Length")
length, err := strconv.ParseInt(lengthHeader, 10, 0)
if err != nil {
return -1, err
}
return int(length), nil
case response.StatusCode == http.StatusNotFound:
return -1, nil
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
err = decoder.Decode(&errors)
if err != nil {
return -1, err
}
return -1, errors
default:
response.Body.Close()
return -1, &registry.UnexpectedHTTPStatusError{Status: response.Status}
}
}
func (r *clientImpl) GetBlob(name, digest string, byteOffset int) (io.ReadCloser, int, error) {
getRequest, err := http.NewRequest("GET",
fmt.Sprintf("%s/v2/%s/layer/%s", r.Endpoint, name, tarsum), nil)
fmt.Sprintf("%s/v2/%s/blob/%s", r.Endpoint, name, digest), nil)
if err != nil {
return nil, 0, err
}
@ -233,9 +267,6 @@ func (r *clientImpl) GetImageLayer(name, tarsum string, byteOffset int) (io.Read
return nil, 0, err
}
if response.StatusCode == http.StatusNotFound {
return nil, 0, &registry.LayerNotFoundError{Name: name, TarSum: tarsum}
}
// TODO(bbland): handle other status codes, like 5xx errors
switch {
case response.StatusCode == http.StatusOK:
@ -247,7 +278,7 @@ func (r *clientImpl) GetImageLayer(name, tarsum string, byteOffset int) (io.Read
return response.Body, int(length), nil
case response.StatusCode == http.StatusNotFound:
response.Body.Close()
return nil, 0, &registry.LayerNotFoundError{Name: name, TarSum: tarsum}
return nil, 0, &registry.BlobNotFoundError{Name: name, Digest: digest}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -262,9 +293,9 @@ func (r *clientImpl) GetImageLayer(name, tarsum string, byteOffset int) (io.Read
}
}
func (r *clientImpl) InitiateLayerUpload(name, tarsum string) (string, error) {
func (r *clientImpl) InitiateBlobUpload(name string) (string, error) {
postRequest, err := http.NewRequest("POST",
fmt.Sprintf("%s/v2/%s/layer/%s/upload/", r.Endpoint, name, tarsum), nil)
fmt.Sprintf("%s/v2/%s/blob/upload/", r.Endpoint, name), nil)
if err != nil {
return "", err
}
@ -279,8 +310,8 @@ func (r *clientImpl) InitiateLayerUpload(name, tarsum string) (string, error) {
switch {
case response.StatusCode == http.StatusAccepted:
return response.Header.Get("Location"), nil
case response.StatusCode == http.StatusNotModified:
return "", &registry.LayerAlreadyExistsError{Name: name, TarSum: tarsum}
// case response.StatusCode == http.StatusNotFound:
// return
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -294,7 +325,7 @@ func (r *clientImpl) InitiateLayerUpload(name, tarsum string) (string, error) {
}
}
func (r *clientImpl) GetLayerUploadStatus(location string) (int, int, error) {
func (r *clientImpl) GetBlobUploadStatus(location string) (int, int, error) {
response, err := http.Get(fmt.Sprintf("%s%s", r.Endpoint, location))
if err != nil {
return 0, 0, err
@ -306,7 +337,7 @@ func (r *clientImpl) GetLayerUploadStatus(location string) (int, int, error) {
case response.StatusCode == http.StatusNoContent:
return parseRangeHeader(response.Header.Get("Range"))
case response.StatusCode == http.StatusNotFound:
return 0, 0, &registry.LayerUploadNotFoundError{Location: location}
return 0, 0, &registry.BlobUploadNotFoundError{Location: location}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -320,18 +351,18 @@ func (r *clientImpl) GetLayerUploadStatus(location string) (int, int, error) {
}
}
func (r *clientImpl) UploadLayer(location string, layer io.ReadCloser, length int, checksum *registry.Checksum) error {
defer layer.Close()
func (r *clientImpl) UploadBlob(location string, blob io.ReadCloser, length int, digest string) error {
defer blob.Close()
putRequest, err := http.NewRequest("PUT",
fmt.Sprintf("%s%s", r.Endpoint, location), layer)
fmt.Sprintf("%s%s", r.Endpoint, location), blob)
if err != nil {
return err
}
queryValues := url.Values{}
queryValues.Set("length", fmt.Sprint(length))
queryValues.Set(checksum.HashAlgorithm, checksum.Sum)
queryValues.Set("digest", digest)
putRequest.URL.RawQuery = queryValues.Encode()
putRequest.Header.Set("Content-Type", "application/octet-stream")
@ -348,7 +379,7 @@ func (r *clientImpl) UploadLayer(location string, layer io.ReadCloser, length in
case response.StatusCode == http.StatusCreated:
return nil
case response.StatusCode == http.StatusNotFound:
return &registry.LayerUploadNotFoundError{Location: location}
return &registry.BlobUploadNotFoundError{Location: location}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -362,11 +393,11 @@ func (r *clientImpl) UploadLayer(location string, layer io.ReadCloser, length in
}
}
func (r *clientImpl) UploadLayerChunk(location string, layerChunk io.ReadCloser, length, startByte int) error {
defer layerChunk.Close()
func (r *clientImpl) UploadBlobChunk(location string, blobChunk io.ReadCloser, length, startByte int) error {
defer blobChunk.Close()
putRequest, err := http.NewRequest("PUT",
fmt.Sprintf("%s%s", r.Endpoint, location), layerChunk)
fmt.Sprintf("%s%s", r.Endpoint, location), blobChunk)
if err != nil {
return err
}
@ -389,17 +420,17 @@ func (r *clientImpl) UploadLayerChunk(location string, layerChunk io.ReadCloser,
case response.StatusCode == http.StatusAccepted:
return nil
case response.StatusCode == http.StatusRequestedRangeNotSatisfiable:
lastValidRange, layerSize, err := parseRangeHeader(response.Header.Get("Range"))
lastValidRange, blobSize, err := parseRangeHeader(response.Header.Get("Range"))
if err != nil {
return err
}
return &registry.LayerUploadInvalidRangeError{
return &registry.BlobUploadInvalidRangeError{
Location: location,
LastValidRange: lastValidRange,
LayerSize: layerSize,
BlobSize: blobSize,
}
case response.StatusCode == http.StatusNotFound:
return &registry.LayerUploadNotFoundError{Location: location}
return &registry.BlobUploadNotFoundError{Location: location}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -413,7 +444,7 @@ func (r *clientImpl) UploadLayerChunk(location string, layerChunk io.ReadCloser,
}
}
func (r *clientImpl) FinishChunkedLayerUpload(location string, length int, checksum *registry.Checksum) error {
func (r *clientImpl) FinishChunkedBlobUpload(location string, length int, digest string) error {
putRequest, err := http.NewRequest("PUT",
fmt.Sprintf("%s%s", r.Endpoint, location), nil)
if err != nil {
@ -422,7 +453,7 @@ func (r *clientImpl) FinishChunkedLayerUpload(location string, length int, check
queryValues := new(url.Values)
queryValues.Set("length", fmt.Sprint(length))
queryValues.Set(checksum.HashAlgorithm, checksum.Sum)
queryValues.Set("digest", digest)
putRequest.URL.RawQuery = queryValues.Encode()
putRequest.Header.Set("Content-Type", "application/octet-stream")
@ -441,7 +472,7 @@ func (r *clientImpl) FinishChunkedLayerUpload(location string, length int, check
case response.StatusCode == http.StatusCreated:
return nil
case response.StatusCode == http.StatusNotFound:
return &registry.LayerUploadNotFoundError{Location: location}
return &registry.BlobUploadNotFoundError{Location: location}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -455,7 +486,7 @@ func (r *clientImpl) FinishChunkedLayerUpload(location string, length int, check
}
}
func (r *clientImpl) CancelLayerUpload(location string) error {
func (r *clientImpl) CancelBlobUpload(location string) error {
deleteRequest, err := http.NewRequest("DELETE",
fmt.Sprintf("%s%s", r.Endpoint, location), nil)
if err != nil {
@ -473,7 +504,7 @@ func (r *clientImpl) CancelLayerUpload(location string) error {
case response.StatusCode == http.StatusNoContent:
return nil
case response.StatusCode == http.StatusNotFound:
return &registry.LayerUploadNotFoundError{Location: location}
return &registry.BlobUploadNotFoundError{Location: location}
case response.StatusCode >= 400 && response.StatusCode < 500:
errors := new(registry.Errors)
decoder := json.NewDecoder(response.Body)
@ -490,7 +521,7 @@ func (r *clientImpl) CancelLayerUpload(location string) error {
// imageManifestURL is a helper method for returning the full url to an image
// manifest
func (r *clientImpl) imageManifestURL(name, tag string) string {
return fmt.Sprintf("%s/v2/%s/image/%s", r.Endpoint, name, tag)
return fmt.Sprintf("%s/v2/%s/manifest/%s", r.Endpoint, name, tag)
}
// parseRangeHeader parses out the offset and length from a returned Range

136
client/client_test.go
View file

@ -13,85 +13,87 @@ import (
"github.com/docker/docker-registry/test"
)
type testLayer struct {
tarSum string
type testBlob struct {
digest string
contents []byte
}
func TestPush(t *testing.T) {
name := "hello/world"
tag := "sometag"
testLayers := []testLayer{
testBlobs := []testBlob{
{
tarSum: "12345",
digest: "12345",
contents: []byte("some contents"),
},
{
tarSum: "98765",
digest: "98765",
contents: []byte("some other contents"),
},
}
uploadLocations := make([]string, len(testLayers))
layers := make([]registry.FSLayer, len(testLayers))
history := make([]registry.ManifestHistory, len(testLayers))
uploadLocations := make([]string, len(testBlobs))
blobs := make([]registry.FSLayer, len(testBlobs))
history := make([]registry.ManifestHistory, len(testBlobs))
for i, layer := range testLayers {
uploadLocations[i] = fmt.Sprintf("/v2/%s/layer/%s/upload-location-%d", name, layer.tarSum, i)
layers[i] = registry.FSLayer{BlobSum: layer.tarSum}
history[i] = registry.ManifestHistory{V1Compatibility: layer.tarSum}
for i, blob := range testBlobs {
// TODO(bbland): this is returning the same location for all uploads,
// because we can't know which blob will get which location.
// It's sort of okay because we're using unique digests, but this needs
// to change at some point.
uploadLocations[i] = fmt.Sprintf("/v2/%s/blob/test-uuid", name)
blobs[i] = registry.FSLayer{BlobSum: blob.digest}
history[i] = registry.ManifestHistory{V1Compatibility: blob.digest}
}
manifest := &registry.ImageManifest{
Name: name,
Tag: tag,
Architecture: "x86",
FSLayers: layers,
FSLayers: blobs,
History: history,
SchemaVersion: 1,
}
manifestBytes, err := json.Marshal(manifest)
layerRequestResponseMappings := make([]test.RequestResponseMapping, 2*len(testLayers))
for i, layer := range testLayers {
layerRequestResponseMappings[2*i] = test.RequestResponseMapping{
blobRequestResponseMappings := make([]test.RequestResponseMapping, 2*len(testBlobs))
for i, blob := range testBlobs {
blobRequestResponseMappings[2*i] = test.RequestResponseMapping{
Request: test.Request{
Method: "POST",
Route: "/v2/" + name + "/layer/" + layer.tarSum + "/upload/",
Route: "/v2/" + name + "/blob/upload/",
},
Responses: []test.Response{
{
StatusCode: http.StatusAccepted,
Headers: http.Header(map[string][]string{
"Location": {uploadLocations[i]},
}),
},
Response: test.Response{
StatusCode: http.StatusAccepted,
Headers: http.Header(map[string][]string{
"Location": {uploadLocations[i]},
}),
},
}
layerRequestResponseMappings[2*i+1] = test.RequestResponseMapping{
blobRequestResponseMappings[2*i+1] = test.RequestResponseMapping{
Request: test.Request{
Method: "PUT",
Route: uploadLocations[i],
Body: layer.contents,
},
Responses: []test.Response{
{
StatusCode: http.StatusCreated,
QueryParams: map[string][]string{
"length": {fmt.Sprint(len(blob.contents))},
"digest": {blob.digest},
},
Body: blob.contents,
},
Response: test.Response{
StatusCode: http.StatusCreated,
},
}
}
handler := test.NewHandler(append(layerRequestResponseMappings, test.RequestResponseMap{
handler := test.NewHandler(append(blobRequestResponseMappings, test.RequestResponseMap{
test.RequestResponseMapping{
Request: test.Request{
Method: "PUT",
Route: "/v2/" + name + "/image/" + tag,
Route: "/v2/" + name + "/manifest/" + tag,
Body: manifestBytes,
},
Responses: []test.Response{
{
StatusCode: http.StatusOK,
},
Response: test.Response{
StatusCode: http.StatusOK,
},
},
}...))
@ -103,8 +105,8 @@ func TestPush(t *testing.T) {
layerStorage: make(map[string]Layer),
}
for _, layer := range testLayers {
l, err := objectStore.Layer(layer.tarSum)
for _, blob := range testBlobs {
l, err := objectStore.Layer(blob.digest)
if err != nil {
t.Fatal(err)
}
@ -114,7 +116,7 @@ func TestPush(t *testing.T) {
t.Fatal(err)
}
writer.Write(layer.contents)
writer.Write(blob.contents)
writer.Close()
}
@ -129,61 +131,57 @@ func TestPush(t *testing.T) {
func TestPull(t *testing.T) {
name := "hello/world"
tag := "sometag"
testLayers := []testLayer{
testBlobs := []testBlob{
{
tarSum: "12345",
digest: "12345",
contents: []byte("some contents"),
},
{
tarSum: "98765",
digest: "98765",
contents: []byte("some other contents"),
},
}
layers := make([]registry.FSLayer, len(testLayers))
history := make([]registry.ManifestHistory, len(testLayers))
blobs := make([]registry.FSLayer, len(testBlobs))
history := make([]registry.ManifestHistory, len(testBlobs))
for i, layer := range testLayers {
layers[i] = registry.FSLayer{BlobSum: layer.tarSum}
history[i] = registry.ManifestHistory{V1Compatibility: layer.tarSum}
for i, blob := range testBlobs {
blobs[i] = registry.FSLayer{BlobSum: blob.digest}
history[i] = registry.ManifestHistory{V1Compatibility: blob.digest}
}
manifest := &registry.ImageManifest{
Name: name,
Tag: tag,
Architecture: "x86",
FSLayers: layers,
FSLayers: blobs,
History: history,
SchemaVersion: 1,
}
manifestBytes, err := json.Marshal(manifest)
layerRequestResponseMappings := make([]test.RequestResponseMapping, len(testLayers))
for i, layer := range testLayers {
layerRequestResponseMappings[i] = test.RequestResponseMapping{
blobRequestResponseMappings := make([]test.RequestResponseMapping, len(testBlobs))
for i, blob := range testBlobs {
blobRequestResponseMappings[i] = test.RequestResponseMapping{
Request: test.Request{
Method: "GET",
Route: "/v2/" + name + "/layer/" + layer.tarSum,
Route: "/v2/" + name + "/blob/" + blob.digest,
},
Responses: []test.Response{
{
StatusCode: http.StatusOK,
Body: layer.contents,
},
Response: test.Response{
StatusCode: http.StatusOK,
Body: blob.contents,
},
}
}
handler := test.NewHandler(append(layerRequestResponseMappings, test.RequestResponseMap{
handler := test.NewHandler(append(blobRequestResponseMappings, test.RequestResponseMap{
test.RequestResponseMapping{
Request: test.Request{
Method: "GET",
Route: "/v2/" + name + "/image/" + tag,
Route: "/v2/" + name + "/manifest/" + tag,
},
Responses: []test.Response{
{
StatusCode: http.StatusOK,
Body: manifestBytes,
},
Response: test.Response{
StatusCode: http.StatusOK,
Body: manifestBytes,
},
},
}...))
@ -214,8 +212,8 @@ func TestPull(t *testing.T) {
t.Fatal("Incorrect manifest")
}
for _, layer := range testLayers {
l, err := objectStore.Layer(layer.tarSum)
for _, blob := range testBlobs {
l, err := objectStore.Layer(blob.digest)
if err != nil {
t.Fatal(err)
}
@ -226,13 +224,13 @@ func TestPull(t *testing.T) {
}
defer reader.Close()
layerBytes, err := ioutil.ReadAll(reader)
blobBytes, err := ioutil.ReadAll(reader)
if err != nil {
t.Fatal(err)
}
if string(layerBytes) != string(layer.contents) {
t.Fatal("Incorrect layer")
if string(blobBytes) != string(blob.contents) {
t.Fatal("Incorrect blob")
}
}
}

2
client/pull.go
View file

@ -99,7 +99,7 @@ func pullLayer(c Client, objectStore ObjectStore, name string, fsLayer registry.
}
defer writer.Close()
layerReader, length, err := c.GetImageLayer(name, fsLayer.BlobSum, 0)
layerReader, length, err := c.GetBlob(name, fsLayer.BlobSum, 0)
if err != nil {
log.WithFields(log.Fields{
"error": err,

46
client/push.go
View file

@ -2,7 +2,6 @@ package client
import (
"bytes"
"crypto/sha1"
"io"
"io/ioutil"
@ -89,25 +88,10 @@ func pushLayer(c Client, objectStore ObjectStore, name string, fsLayer registry.
}).Warn("Unable to read local layer")
return err
}
location, err := c.InitiateLayerUpload(name, fsLayer.BlobSum)
if _, ok := err.(*registry.LayerAlreadyExistsError); ok {
log.WithField("layer", fsLayer).Info("Layer already exists")
return nil
}
if err != nil {
log.WithFields(log.Fields{
"error": err,
"layer": fsLayer,
}).Warn("Unable to upload layer")
return err
}
defer layerReader.Close()
layerBuffer := new(bytes.Buffer)
checksum := sha1.New()
teeReader := io.TeeReader(layerReader, checksum)
_, err = io.Copy(layerBuffer, teeReader)
layerSize, err := io.Copy(layerBuffer, layerReader)
if err != nil {
log.WithFields(log.Fields{
"error": err,
@ -116,9 +100,29 @@ func pushLayer(c Client, objectStore ObjectStore, name string, fsLayer registry.
return err
}
err = c.UploadLayer(location, ioutil.NopCloser(layerBuffer), layerBuffer.Len(),
&registry.Checksum{HashAlgorithm: "sha1", Sum: string(checksum.Sum(nil))},
)
length, err := c.BlobLength(name, fsLayer.BlobSum)
if err != nil {
log.WithFields(log.Fields{
"error": err,
"layer": fsLayer,
}).Warn("Unable to check existence of remote layer")
return err
}
if length >= 0 {
log.WithField("layer", fsLayer).Info("Layer already exists")
return nil
}
location, err := c.InitiateBlobUpload(name)
if err != nil {
log.WithFields(log.Fields{
"error": err,
"layer": fsLayer,
}).Warn("Unable to upload layer")
return err
}
err = c.UploadBlob(location, ioutil.NopCloser(layerBuffer), int(layerSize), fsLayer.BlobSum)
if err != nil {
log.WithFields(log.Fields{
"error": err,