Feedback update
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:
parent
eb2ac4301f
commit
8db2145b81
4 changed files with 68 additions and 62 deletions
|
@ -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 {
|
||||||
|
|
|
@ -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
|
|
||||||
|
|
|
@ -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}
|
||||||
}
|
}
|
||||||
|
|
|
@ -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)
|
||||||
|
|
Loading…
Reference in a new issue