distribution/registry/client/repository.go
Patrick Devine b7e26bac74 Create Repositories method
This change removes the Catalog Service and replaces it with a more
simplistic Repositories() method for obtaining a catalog of all
repositories.  The Repositories method takes a pre-allocated slice
and fills it up to the size of the slice and returns the amount
filled.  The catalog is returned lexicographically and will start
being filled from the last entry passed to Repositories().  If there
are no more entries to fill, io.EOF will be returned.

Signed-off-by: Patrick Devine <patrick.devine@docker.com>

Conflicts:
	registry/client/repository.go
	registry/handlers/api_test.go
2015-07-21 21:45:14 -07:00

517 lines
12 KiB
Go

package client
import (
"bytes"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"net/http"
"net/url"
"strconv"
"time"
"github.com/docker/distribution"
"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/client/transport"
"github.com/docker/distribution/registry/storage/cache"
"github.com/docker/distribution/registry/storage/cache/memory"
)
// NewRepository creates a new Repository for the given repository name and base URL
func NewRepository(ctx context.Context, name, baseURL string, transport http.RoundTripper) (distribution.Repository, error) {
if err := v2.ValidateRepositoryName(name); err != nil {
return nil, err
}
ub, err := v2.NewURLBuilderFromString(baseURL)
if err != nil {
return nil, err
}
client := &http.Client{
Transport: transport,
// TODO(dmcgowan): create cookie jar
}
return &repository{
client: client,
ub: ub,
name: name,
context: ctx,
}, nil
}
type repository struct {
client *http.Client
ub *v2.URLBuilder
context context.Context
name string
}
func (r *repository) Name() string {
return r.name
}
func (r *repository) Blobs(ctx context.Context) distribution.BlobStore {
statter := &blobStatter{
name: r.Name(),
ub: r.ub,
client: r.client,
}
return &blobs{
name: r.Name(),
ub: r.ub,
client: r.client,
statter: cache.NewCachedBlobStatter(memory.NewInMemoryBlobDescriptorCacheProvider(), statter),
}
}
func (r *repository) Manifests(ctx context.Context, options ...distribution.ManifestServiceOption) (distribution.ManifestService, error) {
// todo(richardscothern): options should be sent over the wire
return &manifests{
name: r.Name(),
ub: r.ub,
client: r.client,
etags: make(map[string]string),
}, nil
}
func (r *repository) Signatures() distribution.SignatureService {
ms, _ := r.Manifests(r.context)
return &signatures{
manifests: ms,
}
}
type signatures struct {
manifests distribution.ManifestService
}
func (s *signatures) Get(dgst digest.Digest) ([][]byte, error) {
m, err := s.manifests.Get(dgst)
if err != nil {
return nil, err
}
return m.Signatures()
}
func (s *signatures) Put(dgst digest.Digest, signatures ...[]byte) error {
panic("not implemented")
}
type manifests struct {
name string
ub *v2.URLBuilder
client *http.Client
etags map[string]string
}
func (ms *manifests) Tags() ([]string, error) {
u, err := ms.ub.BuildTagsURL(ms.name)
if err != nil {
return nil, err
}
resp, err := ms.client.Get(u)
if err != nil {
return nil, err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusOK:
b, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, err
}
tagsResponse := struct {
Tags []string `json:"tags"`
}{}
if err := json.Unmarshal(b, &tagsResponse); err != nil {
return nil, err
}
return tagsResponse.Tags, nil
case http.StatusNotFound:
return nil, nil
default:
return nil, handleErrorResponse(resp)
}
}
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())
}
func (ms *manifests) ExistsByTag(tag string) (bool, error) {
u, err := ms.ub.BuildManifestURL(ms.name, tag)
if err != nil {
return false, err
}
resp, err := ms.client.Head(u)
if err != nil {
return false, err
}
switch resp.StatusCode {
case http.StatusOK:
return true, nil
case http.StatusNotFound:
return false, nil
default:
return false, handleErrorResponse(resp)
}
}
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())
}
// AddEtagToTag allows a client to supply an eTag to GetByTag which will
// be used for a conditional HTTP request. If the eTag matches, a nil
// manifest and nil error will be returned.
func AddEtagToTag(tagName, dgst string) distribution.ManifestServiceOption {
return func(ms distribution.ManifestService) error {
if ms, ok := ms.(*manifests); ok {
ms.etags[tagName] = dgst
return nil
}
return fmt.Errorf("etag options is a client-only option")
}
}
func (ms *manifests) GetByTag(tag string, options ...distribution.ManifestServiceOption) (*manifest.SignedManifest, error) {
for _, option := range options {
err := option(ms)
if err != nil {
return nil, err
}
}
u, err := ms.ub.BuildManifestURL(ms.name, tag)
if err != nil {
return nil, err
}
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, err
}
if _, ok := ms.etags[tag]; ok {
req.Header.Set("eTag", ms.etags[tag])
}
resp, err := ms.client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusOK:
var sm manifest.SignedManifest
decoder := json.NewDecoder(resp.Body)
if err := decoder.Decode(&sm); err != nil {
return nil, err
}
return &sm, nil
case http.StatusNotModified:
return nil, nil
default:
return nil, handleErrorResponse(resp)
}
}
func (ms *manifests) Put(m *manifest.SignedManifest) error {
manifestURL, err := ms.ub.BuildManifestURL(ms.name, m.Tag)
if err != nil {
return err
}
// todo(richardscothern): do something with options here when they become applicable
putRequest, err := http.NewRequest("PUT", manifestURL, bytes.NewReader(m.Raw))
if err != nil {
return err
}
resp, err := ms.client.Do(putRequest)
if err != nil {
return err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusAccepted:
// TODO(dmcgowan): make use of digest header
return nil
default:
return handleErrorResponse(resp)
}
}
func (ms *manifests) Delete(dgst digest.Digest) error {
u, err := ms.ub.BuildManifestURL(ms.name, dgst.String())
if err != nil {
return err
}
req, err := http.NewRequest("DELETE", u, nil)
if err != nil {
return err
}
resp, err := ms.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusOK:
return nil
default:
return handleErrorResponse(resp)
}
}
type blobs struct {
name string
ub *v2.URLBuilder
client *http.Client
statter distribution.BlobStatter
}
func sanitizeLocation(location, source string) (string, error) {
locationURL, err := url.Parse(location)
if err != nil {
return "", err
}
if locationURL.Scheme == "" {
sourceURL, err := url.Parse(source)
if err != nil {
return "", err
}
locationURL = &url.URL{
Scheme: sourceURL.Scheme,
Host: sourceURL.Host,
Path: location,
}
location = locationURL.String()
}
return location, nil
}
func (bs *blobs) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
return bs.statter.Stat(ctx, dgst)
}
func (bs *blobs) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
desc, err := bs.Stat(ctx, dgst)
if err != nil {
return nil, err
}
reader, err := bs.Open(ctx, desc.Digest)
if err != nil {
return nil, err
}
defer reader.Close()
return ioutil.ReadAll(reader)
}
func (bs *blobs) Open(ctx context.Context, dgst digest.Digest) (distribution.ReadSeekCloser, error) {
stat, err := bs.statter.Stat(ctx, dgst)
if err != nil {
return nil, err
}
blobURL, err := bs.ub.BuildBlobURL(bs.name, stat.Digest)
if err != nil {
return nil, err
}
return transport.NewHTTPReadSeeker(bs.client, blobURL, stat.Size), nil
}
func (bs *blobs) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
panic("not implemented")
}
func (bs *blobs) Put(ctx context.Context, mediaType string, p []byte) (distribution.Descriptor, error) {
writer, err := bs.Create(ctx)
if err != nil {
return distribution.Descriptor{}, err
}
dgstr := digest.Canonical.New()
n, err := io.Copy(writer, io.TeeReader(bytes.NewReader(p), dgstr.Hash()))
if err != nil {
return distribution.Descriptor{}, err
}
if n < int64(len(p)) {
return distribution.Descriptor{}, fmt.Errorf("short copy: wrote %d of %d", n, len(p))
}
desc := distribution.Descriptor{
MediaType: mediaType,
Size: int64(len(p)),
Digest: dgstr.Digest(),
}
return writer.Commit(ctx, desc)
}
func (bs *blobs) Create(ctx context.Context) (distribution.BlobWriter, error) {
u, err := bs.ub.BuildBlobUploadURL(bs.name)
resp, err := bs.client.Post(u, "", nil)
if err != nil {
return nil, err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusAccepted:
// TODO(dmcgowan): Check for invalid UUID
uuid := resp.Header.Get("Docker-Upload-UUID")
location, err := sanitizeLocation(resp.Header.Get("Location"), u)
if err != nil {
return nil, err
}
return &httpBlobUpload{
statter: bs.statter,
client: bs.client,
uuid: uuid,
startedAt: time.Now(),
location: location,
}, nil
default:
return nil, handleErrorResponse(resp)
}
}
func (bs *blobs) Resume(ctx context.Context, id string) (distribution.BlobWriter, error) {
panic("not implemented")
}
type blobStatter struct {
name string
ub *v2.URLBuilder
client *http.Client
}
func (bs *blobStatter) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
u, err := bs.ub.BuildBlobURL(bs.name, dgst)
if err != nil {
return distribution.Descriptor{}, err
}
resp, err := bs.client.Head(u)
if err != nil {
return distribution.Descriptor{}, err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusOK:
lengthHeader := resp.Header.Get("Content-Length")
length, err := strconv.ParseInt(lengthHeader, 10, 64)
if err != nil {
return distribution.Descriptor{}, fmt.Errorf("error parsing content-length: %v", err)
}
return distribution.Descriptor{
MediaType: resp.Header.Get("Content-Type"),
Size: length,
Digest: dgst,
}, nil
case http.StatusNotFound:
return distribution.Descriptor{}, distribution.ErrBlobUnknown
default:
return distribution.Descriptor{}, handleErrorResponse(resp)
}
}
func buildCatalogValues(maxEntries int, last string) url.Values {
values := url.Values{}
if maxEntries > 0 {
values.Add("n", strconv.Itoa(maxEntries))
}
if last != "" {
values.Add("last", last)
}
return values
}
// Repositories returns a lexigraphically sorted catalog given a base URL. The 'entries' slice will be filled up to the size
// of the slice, starting at the value provided in 'last'. The number of entries will be returned along with io.EOF if there
// are no more entries
func Repositories(ctx context.Context, baseURL string, entries []string, last string, transport http.RoundTripper) (int, error) {
var numFilled int
var returnErr error
ub, err := v2.NewURLBuilderFromString(baseURL)
if err != nil {
return 0, err
}
client := &http.Client{
Transport: transport,
Timeout: 1 * time.Minute,
}
values := buildCatalogValues(len(entries), last)
u, err := ub.BuildCatalogURL(values)
if err != nil {
return 0, err
}
resp, err := client.Get(u)
if err != nil {
return 0, err
}
defer resp.Body.Close()
switch resp.StatusCode {
case http.StatusOK:
var ctlg struct {
Repositories []string `json:"repositories"`
}
decoder := json.NewDecoder(resp.Body)
if err := decoder.Decode(&ctlg); err != nil {
return 0, err
}
for cnt := range ctlg.Repositories {
entries[cnt] = ctlg.Repositories[cnt]
}
numFilled = len(ctlg.Repositories)
link := resp.Header.Get("Link")
if link == "" {
returnErr = io.EOF
}
default:
return 0, handleErrorResponse(resp)
}
return numFilled, returnErr
}