daniel
/
distribution
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #848 from stevvooe/next-generation 

Relax requirement for size argument during blob upload
master
Olivier Gambier 2014-12-13 15:59:04 -08:00
parent 8fd47c1c18 fc7b47cdae
commit 55b49254a0
6 changed files with 129 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
api_test.go
View File

@ -460,6 +460,9 @@ func pushLayer(t *testing.T, ub *v2.URLBuilder, name string, dgst digest.Digest,
u.RawQuery = url.Values{
"digest": []string{dgst.String()},
// TODO(stevvooe): Layer upload can be completed with and without size
// argument. We'll need to add a test that checks the latter path.
"size": []string{fmt.Sprint(rsLength)},
}.Encode()

85
client/client.go
View File

@ -71,19 +71,33 @@ type Client interface {
// 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.
func New(endpoint string) Client {
return &clientImpl{endpoint}
func New(endpoint string) (Client, error) {
ub, err := v2.NewURLBuilderFromString(endpoint)
if err != nil {
return nil, err
}
return &clientImpl{
endpoint: endpoint,
ub: ub,
}, nil
}
// clientImpl is the default implementation of the Client interface
type clientImpl struct {
Endpoint string
endpoint string
ub *v2.URLBuilder
}
// TODO(bbland): use consistent route generation between server and client
func (r *clientImpl) GetImageManifest(name, tag string) (*storage.SignedManifest, error) {
response, err := http.Get(r.imageManifestURL(name, tag))
manifestURL, err := r.ub.BuildManifestURL(name, tag)
if err != nil {
return nil, err
}
response, err := http.Get(manifestURL)
if err != nil {
return nil, err
}
@ -119,8 +133,12 @@ func (r *clientImpl) GetImageManifest(name, tag string) (*storage.SignedManifest
}
func (r *clientImpl) PutImageManifest(name, tag string, manifest *storage.SignedManifest) error {
putRequest, err := http.NewRequest("PUT",
r.imageManifestURL(name, tag), bytes.NewReader(manifest.Raw))
manifestURL, err := r.ub.BuildManifestURL(name, tag)
if err != nil {
return err
}
putRequest, err := http.NewRequest("PUT", manifestURL, bytes.NewReader(manifest.Raw))
if err != nil {
return err
}
@ -150,8 +168,12 @@ func (r *clientImpl) PutImageManifest(name, tag string, manifest *storage.Signed
}
func (r *clientImpl) DeleteImage(name, tag string) error {
deleteRequest, err := http.NewRequest("DELETE",
r.imageManifestURL(name, tag), nil)
manifestURL, err := r.ub.BuildManifestURL(name, tag)
if err != nil {
return err
}
deleteRequest, err := http.NewRequest("DELETE", manifestURL, nil)
if err != nil {
return err
}
@ -184,7 +206,12 @@ func (r *clientImpl) DeleteImage(name, tag string) error {
}
func (r *clientImpl) ListImageTags(name string) ([]string, error) {
response, err := http.Get(fmt.Sprintf("%s/v2/%s/tags/list", r.Endpoint, name))
tagsURL, err := r.ub.BuildTagsURL(name)
if err != nil {
return nil, err
}
response, err := http.Get(tagsURL)
if err != nil {
return nil, err
}
@ -222,7 +249,12 @@ func (r *clientImpl) ListImageTags(name string) ([]string, error) {
}
func (r *clientImpl) BlobLength(name string, dgst digest.Digest) (int, error) {
response, err := http.Head(fmt.Sprintf("%s/v2/%s/blobs/%s", r.Endpoint, name, dgst))
blobURL, err := r.ub.BuildBlobURL(name, dgst)
if err != nil {
return -1, err
}
response, err := http.Head(blobURL)
if err != nil {
return -1, err
}
@ -254,8 +286,12 @@ func (r *clientImpl) BlobLength(name string, dgst digest.Digest) (int, error) {
}
func (r *clientImpl) GetBlob(name string, dgst digest.Digest, byteOffset int) (io.ReadCloser, int, error) {
getRequest, err := http.NewRequest("GET",
fmt.Sprintf("%s/v2/%s/blobs/%s", r.Endpoint, name, dgst), nil)
blobURL, err := r.ub.BuildBlobURL(name, dgst)
if err != nil {
return nil, 0, err
}
getRequest, err := http.NewRequest("GET", blobURL, nil)
if err != nil {
return nil, 0, err
}
@ -293,8 +329,12 @@ func (r *clientImpl) GetBlob(name string, dgst digest.Digest, byteOffset int) (i
}
func (r *clientImpl) InitiateBlobUpload(name string) (string, error) {
postRequest, err := http.NewRequest("POST",
fmt.Sprintf("%s/v2/%s/blobs/uploads/", r.Endpoint, name), nil)
uploadURL, err := r.ub.BuildBlobUploadURL(name)
if err != nil {
return "", err
}
postRequest, err := http.NewRequest("POST", uploadURL, nil)
if err != nil {
return "", err
}
@ -359,7 +399,6 @@ func (r *clientImpl) UploadBlob(location string, blob io.ReadCloser, length int,
}
queryValues := url.Values{}
queryValues.Set("size", fmt.Sprint(length))
queryValues.Set("digest", dgst.String())
putRequest.URL.RawQuery = queryValues.Encode()
@ -394,8 +433,7 @@ func (r *clientImpl) UploadBlob(location string, blob io.ReadCloser, length int,
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), blobChunk)
putRequest, err := http.NewRequest("PUT", location, blobChunk)
if err != nil {
return err
}
@ -443,14 +481,12 @@ func (r *clientImpl) UploadBlobChunk(location string, blobChunk io.ReadCloser, l
}
func (r *clientImpl) FinishChunkedBlobUpload(location string, length int, dgst digest.Digest) error {
putRequest, err := http.NewRequest("PUT",
fmt.Sprintf("%s%s", r.Endpoint, location), nil)
putRequest, err := http.NewRequest("PUT", location, nil)
if err != nil {
return err
}
queryValues := new(url.Values)
queryValues.Set("size", fmt.Sprint(length))
queryValues.Set("digest", dgst.String())
putRequest.URL.RawQuery = queryValues.Encode()
@ -485,8 +521,7 @@ func (r *clientImpl) FinishChunkedBlobUpload(location string, length int, dgst d
}
func (r *clientImpl) CancelBlobUpload(location string) error {
deleteRequest, err := http.NewRequest("DELETE",
fmt.Sprintf("%s%s", r.Endpoint, location), nil)
deleteRequest, err := http.NewRequest("DELETE", location, nil)
if err != nil {
return err
}
@ -516,12 +551,6 @@ func (r *clientImpl) CancelBlobUpload(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/manifests/%s", r.Endpoint, name, tag)
}
// parseRangeHeader parses out the offset and length from a returned Range
// header
func parseRangeHeader(byteRangeHeader string) (int, int, error) {

28
client/client_test.go
View File

@ -24,11 +24,11 @@ func TestPush(t *testing.T) {
tag := "sometag"
testBlobs := []testBlob{
{
digest: "12345",
digest: "tarsum.v2+sha256:12345",
contents: []byte("some contents"),
},
{
digest: "98765",
digest: "tarsum.v2+sha256:98765",
contents: []byte("some other contents"),
},
}
@ -80,7 +80,6 @@ func TestPush(t *testing.T) {
Method: "PUT",
Route: uploadLocations[i],
QueryParams: map[string][]string{
"length": {fmt.Sprint(len(blob.contents))},
"digest": {blob.digest.String()},
},
Body: blob.contents,
@ -114,7 +113,10 @@ func TestPush(t *testing.T) {
})
server = httptest.NewServer(hack)
client := New(server.URL)
client, err := New(server.URL)
if err != nil {
t.Fatalf("error creating client: %v", err)
}
objectStore := &memoryObjectStore{
mutex: new(sync.Mutex),
manifestStorage: make(map[string]*storage.SignedManifest),
@ -150,11 +152,11 @@ func TestPull(t *testing.T) {
tag := "sometag"
testBlobs := []testBlob{
{
digest: "12345",
digest: "tarsum.v2+sha256:12345",
contents: []byte("some contents"),
},
{
digest: "98765",
digest: "tarsum.v2+sha256:98765",
contents: []byte("some other contents"),
},
}
@ -205,7 +207,10 @@ func TestPull(t *testing.T) {
},
}))
server := httptest.NewServer(handler)
client := New(server.URL)
client, err := New(server.URL)
if err != nil {
t.Fatalf("error creating client: %v", err)
}
objectStore := &memoryObjectStore{
mutex: new(sync.Mutex),
manifestStorage: make(map[string]*storage.SignedManifest),
@ -259,11 +264,11 @@ func TestPullResume(t *testing.T) {
tag := "sometag"
testBlobs := []testBlob{
{
digest: "12345",
digest: "tarsum.v2+sha256:12345",
contents: []byte("some contents"),
},
{
digest: "98765",
digest: "tarsum.v2+sha256:98765",
contents: []byte("some other contents"),
},
}
@ -329,7 +334,10 @@ func TestPullResume(t *testing.T) {
handler := testutil.NewHandler(layerRequestResponseMappings)
server := httptest.NewServer(handler)
client := New(server.URL)
client, err := New(server.URL)
if err != nil {
t.Fatalf("error creating client: %v", err)
}
objectStore := &memoryObjectStore{
mutex: new(sync.Mutex),
manifestStorage: make(map[string]*storage.SignedManifest),

20
layerupload.go
View File

@ -119,11 +119,22 @@ func (luh *layerUploadHandler) PutLayerChunk(w http.ResponseWriter, r *http.Requ
if err := luh.maybeCompleteUpload(w, r); err != nil {
if err != errNotReadyToComplete {
switch err := err.(type) {
case storage.ErrLayerInvalidSize:
w.WriteHeader(http.StatusBadRequest)
luh.Errors.Push(v2.ErrorCodeSizeInvalid, err)
return
case storage.ErrLayerInvalidDigest:
w.WriteHeader(http.StatusBadRequest)
luh.Errors.Push(v2.ErrorCodeDigestInvalid, err)
return
default:
w.WriteHeader(http.StatusInternalServerError)
luh.Errors.Push(v2.ErrorCodeUnknown, err)
return
}
}
}
if err := luh.layerUploadResponse(w, r); err != nil {
w.WriteHeader(http.StatusInternalServerError) // Error conditions here?
@ -173,7 +184,7 @@ func (luh *layerUploadHandler) maybeCompleteUpload(w http.ResponseWriter, r *htt
dgstStr := r.FormValue("digest") // TODO(stevvooe): Support multiple digest parameters!
sizeStr := r.FormValue("size")
if dgstStr == "" || sizeStr == "" {
if dgstStr == "" {
return errNotReadyToComplete
}
@ -182,10 +193,15 @@ func (luh *layerUploadHandler) maybeCompleteUpload(w http.ResponseWriter, r *htt
return err
}
size, err := strconv.ParseInt(sizeStr, 10, 64)
var size int64
if sizeStr != "" {
size, err = strconv.ParseInt(sizeStr, 10, 64)
if err != nil {
return err
}
} else {
size = -1
}
luh.completeUpload(w, r, size, dgst)
return nil

19
storage/layer.go
View File

@ -43,9 +43,14 @@ type LayerUpload interface {
// Offset returns the position of the last byte written to this layer.
Offset() int64
// TODO(stevvooe): Consider completely removing the size check from this
// interface. The digest check may be adequate and we are making it
// optional in the HTTP API.
// Finish marks the upload as completed, returning a valid handle to the
// uploaded layer. The final size and digest are validated against the
// contents of the uploaded layer.
// contents of the uploaded layer. If the size is negative, only the
// digest will be checked.
Finish(size int64, digest digest.Digest) (Layer, error)
// Cancel the layer upload process.
@ -62,9 +67,6 @@ var (
// ErrLayerUploadUnknown returned when upload is not found.
ErrLayerUploadUnknown = fmt.Errorf("layer upload unknown")
// ErrLayerInvalidLength returned when length check fails.
ErrLayerInvalidLength = fmt.Errorf("invalid layer length")
// ErrLayerClosed returned when an operation is attempted on a closed
// Layer or LayerUpload.
ErrLayerClosed = fmt.Errorf("layer closed")
@ -87,3 +89,12 @@ type ErrLayerInvalidDigest struct {
func (err ErrLayerInvalidDigest) Error() string {
return fmt.Sprintf("invalid digest for referenced layer: %v", err.FSLayer.BlobSum)
}
// ErrLayerInvalidSize returned when length check fails.
type ErrLayerInvalidSize struct {
Size int64
}
func (err ErrLayerInvalidSize) Error() string {
return fmt.Sprintf("invalid layer size: %d", err.Size)
}

19
storage/layerupload.go
View File

@ -110,7 +110,7 @@ func (luc *layerUploadController) Finish(size int64, digest digest.Digest) (Laye
if nn, err := luc.writeLayer(fp, digest); err != nil {
// Cleanup?
return nil, err
} else if nn != size {
} else if size >= 0 && nn != size {
// TODO(stevvooe): Short write. Will have to delete the location and
// report an error. This error needs to be reported to the client.
return nil, fmt.Errorf("short write writing layer")
@ -252,9 +252,10 @@ func (luc *layerUploadController) validateLayer(fp layerFile, size int64, dgst d
return "", err
}
if end != size {
// Only check size if it is greater than
if size >= 0 && end != size {
// Fast path length check.
return "", ErrLayerInvalidLength
return "", ErrLayerInvalidSize{Size: size}
}
// Now seek back to start and take care of the digest.
@ -262,8 +263,12 @@ func (luc *layerUploadController) validateLayer(fp layerFile, size int64, dgst d
return "", err
}
tr := io.TeeReader(fp, lengthVerifier)
tr = io.TeeReader(tr, digestVerifier)
tr := io.TeeReader(fp, digestVerifier)
// Only verify the size if a positive size argument has been passed.
if size >= 0 {
tr = io.TeeReader(tr, lengthVerifier)
}
// TODO(stevvooe): This is one of the places we need a Digester write
// sink. Instead, its read driven. This migth be okay.
@ -274,8 +279,8 @@ func (luc *layerUploadController) validateLayer(fp layerFile, size int64, dgst d
return "", err
}
if !lengthVerifier.Verified() {
return "", ErrLayerInvalidLength
if size >= 0 && !lengthVerifier.Verified() {
return "", ErrLayerInvalidSize{Size: size}
}
if !digestVerifier.Verified() {