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

Feedback update

Browse source 
Update comments and TODOs
Fix switch style
Updated parse http response to take in reader
Add Cancel implementation
Update blobstore variable name

Signed-off-by: Derek McGowan <derek@mcgstyle.net> (github: dmcgowan)
This commit is contained in:
Derek McGowan 2015-05-19 19:18:30 -07:00
parent 006ddd8283
commit a3276fcc5b
4 changed files with 68 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
registry/client/blob_writer.go
View file

@ -2,7 +2,6 @@ package client
import ( import (
"bytes" "bytes"
"errors"
"fmt" "fmt"
"io" "io"
"io/ioutil" "io/ioutil"
@ -49,7 +48,6 @@ func (hbu *httpBlobUpload) ReadFrom(r io.Reader) (n int64, err error) {
return 0, hbu.handleErrorResponse(resp) return 0, hbu.handleErrorResponse(resp)
} }
// TODO(dmcgowan): Validate headers
hbu.uuid = resp.Header.Get("Docker-Upload-UUID") hbu.uuid = resp.Header.Get("Docker-Upload-UUID")
hbu.location, err = sanitizeLocation(resp.Header.Get("Location"), hbu.location) hbu.location, err = sanitizeLocation(resp.Header.Get("Location"), hbu.location)
if err != nil { if err != nil {
@ -85,7 +83,6 @@ func (hbu *httpBlobUpload) Write(p []byte) (n int, err error) {
return 0, hbu.handleErrorResponse(resp) return 0, hbu.handleErrorResponse(resp)
} }
// TODO(dmcgowan): Validate headers
hbu.uuid = resp.Header.Get("Docker-Upload-UUID") hbu.uuid = resp.Header.Get("Docker-Upload-UUID")
hbu.location, err = sanitizeLocation(resp.Header.Get("Location"), hbu.location) hbu.location, err = sanitizeLocation(resp.Header.Get("Location"), hbu.location)
if err != nil { if err != nil {
@ -110,7 +107,7 @@ func (hbu *httpBlobUpload) Seek(offset int64, whence int) (int64, error) {
case os.SEEK_CUR: case os.SEEK_CUR:
newOffset += int64(offset) newOffset += int64(offset)
case os.SEEK_END: case os.SEEK_END:
return newOffset, errors.New("Cannot seek from end on incomplete upload") newOffset += int64(offset)
case os.SEEK_SET: case os.SEEK_SET:
newOffset = int64(offset) newOffset = int64(offset)
} }
@ -143,6 +140,7 @@ func (hbu *httpBlobUpload) Commit(ctx context.Context, desc distribution.Descrip
if err != nil { if err != nil {
return distribution.Descriptor{}, err return distribution.Descriptor{}, err
} }
defer resp.Body.Close()
if resp.StatusCode != http.StatusCreated { if resp.StatusCode != http.StatusCreated {
return distribution.Descriptor{}, hbu.handleErrorResponse(resp) return distribution.Descriptor{}, hbu.handleErrorResponse(resp)
@ -152,7 +150,22 @@ func (hbu *httpBlobUpload) Commit(ctx context.Context, desc distribution.Descrip
} }
func (hbu *httpBlobUpload) Cancel(ctx context.Context) error { func (hbu *httpBlobUpload) Cancel(ctx context.Context) error {
panic("not implemented") req, err := http.NewRequest("DELETE", hbu.location, nil)
if err != nil {
return err
}
resp, err := hbu.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusNoContent, http.StatusNotFound:
return nil
default:
return hbu.handleErrorResponse(resp)
}
} }
func (hbu *httpBlobUpload) Close() error { func (hbu *httpBlobUpload) Close() error {

10
registry/client/blob_writer_test.go
View file

@ -205,13 +205,3 @@ func TestUploadReadFrom(t *testing.T) {
t.Fatalf("Unexpected response status: %s, expected %s", uploadErr.Status, expected) t.Fatalf("Unexpected response status: %s, expected %s", uploadErr.Status, expected)
} }
} }
//repo distribution.Repository
//client *http.Client
//uuid string
//startedAt time.Time
//location string // always the last value of the location header.
//offset int64
//closed bool

7
registry/client/errors.go
View file

@ -3,6 +3,7 @@ package client
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
@ -34,9 +35,9 @@ func (e *UnexpectedHTTPResponseError) Error() string {
return fmt.Sprintf("Error parsing HTTP response: %s: %q", e.ParseErr.Error(), shortenedResponse) return fmt.Sprintf("Error parsing HTTP response: %s: %q", e.ParseErr.Error(), shortenedResponse)
} }
func parseHTTPErrorResponse(response *http.Response) error { func parseHTTPErrorResponse(r io.Reader) error {
var errors v2.Errors var errors v2.Errors
body, err := ioutil.ReadAll(response.Body) body, err := ioutil.ReadAll(r)
if err != nil { if err != nil {
return err return err
} }
@ -52,7 +53,7 @@ func parseHTTPErrorResponse(response *http.Response) error {
func handleErrorResponse(resp *http.Response) error { func handleErrorResponse(resp *http.Response) error {
if resp.StatusCode >= 400 && resp.StatusCode < 500 { if resp.StatusCode >= 400 && resp.StatusCode < 500 {
return parseHTTPErrorResponse(resp) return parseHTTPErrorResponse(resp.Body)
} }
return &UnexpectedHTTPStatusError{Status: resp.Status} return &UnexpectedHTTPStatusError{Status: resp.Status}
} }

90
registry/client/repository.go
View file

@ -11,12 +11,10 @@ import (
"strconv" "strconv"
"time" "time"
"github.com/docker/distribution/manifest"
"github.com/docker/distribution/digest"
"github.com/docker/distribution" "github.com/docker/distribution"
"github.com/docker/distribution/context" "github.com/docker/distribution/context"
"github.com/docker/distribution/digest"
"github.com/docker/distribution/manifest"
"github.com/docker/distribution/registry/api/v2" "github.com/docker/distribution/registry/api/v2"
"github.com/docker/distribution/registry/client/transport" "github.com/docker/distribution/registry/client/transport"
"github.com/docker/distribution/registry/storage/cache" "github.com/docker/distribution/registry/storage/cache"
@ -108,8 +106,8 @@ func (ms *manifests) Tags() ([]string, error) {
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusOK: case http.StatusOK:
b, err := ioutil.ReadAll(resp.Body) b, err := ioutil.ReadAll(resp.Body)
if err != nil { if err != nil {
return nil, err return nil, err
@ -123,7 +121,7 @@ func (ms *manifests) Tags() ([]string, error) {
} }
return tagsResponse.Tags, nil return tagsResponse.Tags, nil
case resp.StatusCode == http.StatusNotFound: case http.StatusNotFound:
return nil, nil return nil, nil
default: default:
return nil, handleErrorResponse(resp) return nil, handleErrorResponse(resp)
@ -131,6 +129,8 @@ func (ms *manifests) Tags() ([]string, error) {
} }
func (ms *manifests) Exists(dgst digest.Digest) (bool, error) { func (ms *manifests) Exists(dgst digest.Digest) (bool, error) {
// Call by Tag endpoint since the API uses the same
// URL endpoint for tags and digests.
return ms.ExistsByTag(dgst.String()) return ms.ExistsByTag(dgst.String())
} }
@ -145,10 +145,10 @@ func (ms *manifests) ExistsByTag(tag string) (bool, error) {
return false, err return false, err
} }
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusOK: case http.StatusOK:
return true, nil return true, nil
case resp.StatusCode == http.StatusNotFound: case http.StatusNotFound:
return false, nil return false, nil
default: default:
return false, handleErrorResponse(resp) return false, handleErrorResponse(resp)
@ -156,6 +156,8 @@ func (ms *manifests) ExistsByTag(tag string) (bool, error) {
} }
func (ms *manifests) Get(dgst digest.Digest) (*manifest.SignedManifest, error) { func (ms *manifests) Get(dgst digest.Digest) (*manifest.SignedManifest, error) {
// Call by Tag endpoint since the API uses the same
// URL endpoint for tags and digests.
return ms.GetByTag(dgst.String()) return ms.GetByTag(dgst.String())
} }
@ -171,8 +173,8 @@ func (ms *manifests) GetByTag(tag string) (*manifest.SignedManifest, error) {
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusOK: case http.StatusOK:
var sm manifest.SignedManifest var sm manifest.SignedManifest
decoder := json.NewDecoder(resp.Body) decoder := json.NewDecoder(resp.Body)
@ -203,9 +205,9 @@ func (ms *manifests) Put(m *manifest.SignedManifest) error {
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusAccepted: case http.StatusAccepted:
// TODO(dmcgowan): Use or check digest header // TODO(dmcgowan): make use of digest header
return nil return nil
default: default:
return handleErrorResponse(resp) return handleErrorResponse(resp)
@ -228,8 +230,8 @@ func (ms *manifests) Delete(dgst digest.Digest) error {
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusOK: case http.StatusOK:
return nil return nil
default: default:
return handleErrorResponse(resp) return handleErrorResponse(resp)
@ -263,17 +265,17 @@ func sanitizeLocation(location, source string) (string, error) {
return location, nil return location, nil
} }
func (ls *blobs) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { func (bs *blobs) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
return ls.statter.Stat(ctx, dgst) return bs.statter.Stat(ctx, dgst)
} }
func (ls *blobs) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) { func (bs *blobs) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
desc, err := ls.Stat(ctx, dgst) desc, err := bs.Stat(ctx, dgst)
if err != nil { if err != nil {
return nil, err return nil, err
} }
reader, err := ls.Open(ctx, desc.Digest) reader, err := bs.Open(ctx, desc.Digest)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -282,26 +284,26 @@ func (ls *blobs) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
return ioutil.ReadAll(reader) return ioutil.ReadAll(reader)
} }
func (ls *blobs) Open(ctx context.Context, dgst digest.Digest) (distribution.ReadSeekCloser, error) { func (bs *blobs) Open(ctx context.Context, dgst digest.Digest) (distribution.ReadSeekCloser, error) {
stat, err := ls.statter.Stat(ctx, dgst) stat, err := bs.statter.Stat(ctx, dgst)
if err != nil { if err != nil {
return nil, err return nil, err
} }
blobURL, err := ls.ub.BuildBlobURL(ls.Name(), stat.Digest) blobURL, err := bs.ub.BuildBlobURL(bs.Name(), stat.Digest)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return transport.NewHTTPReadSeeker(ls.repository.client, blobURL, stat.Length), nil return transport.NewHTTPReadSeeker(bs.repository.client, blobURL, stat.Length), nil
} }
func (ls *blobs) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error { func (bs *blobs) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
return nil panic("not implemented")
} }
func (ls *blobs) Put(ctx context.Context, mediaType string, p []byte) (distribution.Descriptor, error) { func (bs *blobs) Put(ctx context.Context, mediaType string, p []byte) (distribution.Descriptor, error) {
writer, err := ls.Create(ctx) writer, err := bs.Create(ctx)
if err != nil { if err != nil {
return distribution.Descriptor{}, err return distribution.Descriptor{}, err
} }
@ -323,17 +325,17 @@ func (ls *blobs) Put(ctx context.Context, mediaType string, p []byte) (distribut
return writer.Commit(ctx, desc) return writer.Commit(ctx, desc)
} }
func (ls *blobs) Create(ctx context.Context) (distribution.BlobWriter, error) { func (bs *blobs) Create(ctx context.Context) (distribution.BlobWriter, error) {
u, err := ls.ub.BuildBlobUploadURL(ls.name) u, err := bs.ub.BuildBlobUploadURL(bs.name)
resp, err := ls.client.Post(u, "", nil) resp, err := bs.client.Post(u, "", nil)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusAccepted: case http.StatusAccepted:
// TODO(dmcgowan): Check for invalid UUID // TODO(dmcgowan): Check for invalid UUID
uuid := resp.Header.Get("Docker-Upload-UUID") uuid := resp.Header.Get("Docker-Upload-UUID")
location, err := sanitizeLocation(resp.Header.Get("Location"), u) location, err := sanitizeLocation(resp.Header.Get("Location"), u)
@ -342,8 +344,8 @@ func (ls *blobs) Create(ctx context.Context) (distribution.BlobWriter, error) {
} }
return &httpBlobUpload{ return &httpBlobUpload{
repo: ls.repository, repo: bs.repository,
client: ls.client, client: bs.client,
uuid: uuid, uuid: uuid,
startedAt: time.Now(), startedAt: time.Now(),
location: location, location: location,
@ -353,7 +355,7 @@ func (ls *blobs) Create(ctx context.Context) (distribution.BlobWriter, error) {
} }
} }
func (ls *blobs) Resume(ctx context.Context, id string) (distribution.BlobWriter, error) { func (bs *blobs) Resume(ctx context.Context, id string) (distribution.BlobWriter, error) {
panic("not implemented") panic("not implemented")
} }
@ -361,20 +363,20 @@ type blobStatter struct {
*repository *repository
} }
func (ls *blobStatter) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { func (bs *blobStatter) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
u, err := ls.ub.BuildBlobURL(ls.name, dgst) u, err := bs.ub.BuildBlobURL(bs.name, dgst)
if err != nil { if err != nil {
return distribution.Descriptor{}, err return distribution.Descriptor{}, err
} }
resp, err := ls.client.Head(u) resp, err := bs.client.Head(u)
if err != nil { if err != nil {
return distribution.Descriptor{}, err return distribution.Descriptor{}, err
} }
defer resp.Body.Close() defer resp.Body.Close()
switch { switch resp.StatusCode {
case resp.StatusCode == http.StatusOK: case http.StatusOK:
lengthHeader := resp.Header.Get("Content-Length") lengthHeader := resp.Header.Get("Content-Length")
length, err := strconv.ParseInt(lengthHeader, 10, 64) length, err := strconv.ParseInt(lengthHeader, 10, 64)
if err != nil { if err != nil {
@ -386,7 +388,7 @@ func (ls *blobStatter) Stat(ctx context.Context, dgst digest.Digest) (distributi
Length: length, Length: length,
Digest: dgst, Digest: dgst,
}, nil }, nil
case resp.StatusCode == http.StatusNotFound: case http.StatusNotFound:
return distribution.Descriptor{}, distribution.ErrBlobUnknown return distribution.Descriptor{}, distribution.ErrBlobUnknown
default: default:
return distribution.Descriptor{}, handleErrorResponse(resp) return distribution.Descriptor{}, handleErrorResponse(resp)