Refactor Blob Service API
This PR refactors the blob service API to be oriented around blob descriptors. Identified by digests, blobs become an abstract entity that can be read and written using a descriptor as a handle. This allows blobs to take many forms, such as a ReadSeekCloser or a simple byte buffer, allowing blob oriented operations to better integrate with blob agnostic APIs (such as the `io` package). The error definitions are now better organized to reflect conditions that can only be seen when interacting with the blob API. The main benefit of this is to separate the much smaller metadata from large file storage. Many benefits also follow from this. Reading and writing has been separated into discrete services. Backend implementation is also simplified, by reducing the amount of metadata that needs to be picked up to simply serve a read. This also improves cacheability. "Opening" a blob simply consists of an access check (Stat) and a path calculation. Caching is greatly simplified and we've made the mapping of provisional to canonical hashes a first-class concept. BlobDescriptorService and BlobProvider can be combined in different ways to achieve varying effects. Recommend Review Approach ------------------------- This is a very large patch. While apologies are in order, we are getting a considerable amount of refactoring. Most changes follow from the changes to the root package (distribution), so start there. From there, the main changes are in storage. Looking at (*repository).Blobs will help to understand the how the linkedBlobStore is wired. One can explore the internals within and also branch out into understanding the changes to the caching layer. Following the descriptions below will also help to guide you. To reduce the chances for regressions, it was critical that major changes to unit tests were avoided. Where possible, they are left untouched and where not, the spirit is hopefully captured. Pay particular attention to where behavior may have changed. Storage ------- The primary changes to the `storage` package, other than the interface updates, were to merge the layerstore and blobstore. Blob access is now layered even further. The first layer, blobStore, exposes a global `BlobStatter` and `BlobProvider`. Operations here provide a fast path for most read operations that don't take access control into account. The `linkedBlobStore` layers on top of the `blobStore`, providing repository- scoped blob link management in the backend. The `linkedBlobStore` implements the full `BlobStore` suite, providing access-controlled, repository-local blob writers. The abstraction between the two is slightly broken in that `linkedBlobStore` is the only channel under which one can write into the global blob store. The `linkedBlobStore` also provides flexibility in that it can act over different link sets depending on configuration. This allows us to use the same code for signature links, manifest links and blob links. Eventually, we will fully consolidate this storage. The improved cache flow comes from the `linkedBlobStatter` component of `linkedBlobStore`. Using a `cachedBlobStatter`, these combine together to provide a simple cache hierarchy that should streamline access checks on read and write operations, or at least provide a single path to optimize. The metrics have been changed in a slightly incompatible way since the former operations, Fetch and Exists, are no longer relevant. The fileWriter and fileReader have been slightly modified to support the rest of the changes. The most interesting is the removal of the `Stat` call from `newFileReader`. This was the source of unnecessary round trips that were only present to look up the size of the resulting reader. Now, one must simply pass in the size, requiring the caller to decide whether or not the `Stat` call is appropriate. In several cases, it turned out the caller already had the size already. The `WriterAt` implementation has been removed from `fileWriter`, since it is no longer required for `BlobWriter`, reducing the number of paths which writes may take. Cache ----- Unfortunately, the `cache` package required a near full rewrite. It was pretty mechanical in that the cache is oriented around the `BlobDescriptorService` slightly modified to include the ability to set the values for individual digests. While the implementation is oriented towards caching, it can act as a primary store. Provisions are in place to have repository local metadata, in addition to global metadata. Fallback is implemented as a part of the storage package to maintain this flexibility. One unfortunate side-effect is that caching is now repository-scoped, rather than global. This should have little effect on performance but may increase memory usage. Handlers -------- The `handlers` package has been updated to leverage the new API. For the most part, the changes are superficial or mechanical based on the API changes. This did expose a bug in the handling of provisional vs canonical digests that was fixed in the unit tests. Configuration ------------- One user-facing change has been made to the configuration and is updated in the associated documentation. The `layerinfo` cache parameter has been deprecated by the `blobdescriptor` cache parameter. Both are equivalent and configuration files should be backward compatible. Notifications ------------- Changes the `notification` package are simply to support the interface changes. Context ------- A small change has been made to the tracing log-level. Traces have been moved from "info" to "debug" level to reduce output when not needed. Signed-off-by: Stephen J Day <stephen.day@docker.com>
This commit is contained in:
Stephen J Day
parent
dc348d720b
commit
08401cfdd6
44 changed files with 2426 additions and 2270 deletions
106
docs/storage/cache/cache.go
vendored
106
docs/storage/cache/cache.go
vendored
|
|
@ -1,98 +1,38 @@
|
|||
// Package cache provides facilities to speed up access to the storage
|
||||
// backend. Typically cache implementations deal with internal implementation
|
||||
// details at the backend level, rather than generalized caches for
|
||||
// distribution related interfaces. In other words, unless the cache is
|
||||
// specific to the storage package, it belongs in another package.
|
||||
// backend.
|
||||
package cache
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/digest"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
||||
// ErrNotFound is returned when a meta item is not found.
|
||||
var ErrNotFound = fmt.Errorf("not found")
|
||||
// BlobDescriptorCacheProvider provides repository scoped
|
||||
// BlobDescriptorService cache instances and a global descriptor cache.
|
||||
type BlobDescriptorCacheProvider interface {
|
||||
distribution.BlobDescriptorService
|
||||
|
||||
// LayerMeta describes the backend location and length of layer data.
|
||||
type LayerMeta struct {
|
||||
Path string
|
||||
Length int64
|
||||
RepositoryScoped(repo string) (distribution.BlobDescriptorService, error)
|
||||
}
|
||||
|
||||
// LayerInfoCache is a driver-aware cache of layer metadata. Basically, it
|
||||
// provides a fast cache for checks against repository metadata, avoiding
|
||||
// round trips to backend storage. Note that this is different from a pure
|
||||
// layer cache, which would also provide access to backing data, as well. Such
|
||||
// a cache should be implemented as a middleware, rather than integrated with
|
||||
// the storage backend.
|
||||
//
|
||||
// Note that most implementations rely on the caller to do strict checks on on
|
||||
// repo and dgst arguments, since these are mostly used behind existing
|
||||
// implementations.
|
||||
type LayerInfoCache interface {
|
||||
// Contains returns true if the repository with name contains the layer.
|
||||
Contains(ctx context.Context, repo string, dgst digest.Digest) (bool, error)
|
||||
|
||||
// Add includes the layer in the given repository cache.
|
||||
Add(ctx context.Context, repo string, dgst digest.Digest) error
|
||||
|
||||
// Meta provides the location of the layer on the backend and its size. Membership of a
|
||||
// repository should be tested before using the result, if required.
|
||||
Meta(ctx context.Context, dgst digest.Digest) (LayerMeta, error)
|
||||
|
||||
// SetMeta sets the meta data for the given layer.
|
||||
SetMeta(ctx context.Context, dgst digest.Digest, meta LayerMeta) error
|
||||
func validateDigest(dgst digest.Digest) error {
|
||||
return dgst.Validate()
|
||||
}
|
||||
|
||||
// base implements common checks between cache implementations. Note that
|
||||
// these are not full checks of input, since that should be done by the
|
||||
// caller.
|
||||
type base struct {
|
||||
LayerInfoCache
|
||||
}
|
||||
|
||||
func (b *base) Contains(ctx context.Context, repo string, dgst digest.Digest) (bool, error) {
|
||||
if repo == "" {
|
||||
return false, fmt.Errorf("cache: cannot check for empty repository name")
|
||||
}
|
||||
|
||||
if dgst == "" {
|
||||
return false, fmt.Errorf("cache: cannot check for empty digests")
|
||||
}
|
||||
|
||||
return b.LayerInfoCache.Contains(ctx, repo, dgst)
|
||||
}
|
||||
|
||||
func (b *base) Add(ctx context.Context, repo string, dgst digest.Digest) error {
|
||||
if repo == "" {
|
||||
return fmt.Errorf("cache: cannot add empty repository name")
|
||||
}
|
||||
|
||||
if dgst == "" {
|
||||
return fmt.Errorf("cache: cannot add empty digest")
|
||||
}
|
||||
|
||||
return b.LayerInfoCache.Add(ctx, repo, dgst)
|
||||
}
|
||||
|
||||
func (b *base) Meta(ctx context.Context, dgst digest.Digest) (LayerMeta, error) {
|
||||
if dgst == "" {
|
||||
return LayerMeta{}, fmt.Errorf("cache: cannot get meta for empty digest")
|
||||
}
|
||||
|
||||
return b.LayerInfoCache.Meta(ctx, dgst)
|
||||
}
|
||||
|
||||
func (b *base) SetMeta(ctx context.Context, dgst digest.Digest, meta LayerMeta) error {
|
||||
if dgst == "" {
|
||||
return fmt.Errorf("cache: cannot set meta for empty digest")
|
||||
}
|
||||
|
||||
if meta.Path == "" {
|
||||
return fmt.Errorf("cache: cannot set empty path for meta")
|
||||
}
|
||||
|
||||
return b.LayerInfoCache.SetMeta(ctx, dgst, meta)
|
||||
func validateDescriptor(desc distribution.Descriptor) error {
|
||||
if err := validateDigest(desc.Digest); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if desc.Length < 0 {
|
||||
return fmt.Errorf("cache: invalid length in descriptor: %v < 0", desc.Length)
|
||||
}
|
||||
|
||||
if desc.MediaType == "" {
|
||||
return fmt.Errorf("cache: empty mediatype on descriptor: %v", desc)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
179
docs/storage/cache/cache_test.go
vendored
179
docs/storage/cache/cache_test.go
vendored
|
|
@ -3,84 +3,139 @@ package cache
|
|||
import (
|
||||
"testing"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/context"
|
||||
"github.com/docker/distribution/digest"
|
||||
)
|
||||
|
||||
// checkLayerInfoCache takes a cache implementation through a common set of
|
||||
// operations. If adding new tests, please add them here so new
|
||||
// checkBlobDescriptorCache takes a cache implementation through a common set
|
||||
// of operations. If adding new tests, please add them here so new
|
||||
// implementations get the benefit.
|
||||
func checkLayerInfoCache(t *testing.T, lic LayerInfoCache) {
|
||||
func checkBlobDescriptorCache(t *testing.T, provider BlobDescriptorCacheProvider) {
|
||||
ctx := context.Background()
|
||||
|
||||
exists, err := lic.Contains(ctx, "", "fake:abc")
|
||||
checkBlobDescriptorCacheEmptyRepository(t, ctx, provider)
|
||||
checkBlobDescriptorCacheSetAndRead(t, ctx, provider)
|
||||
}
|
||||
|
||||
func checkBlobDescriptorCacheEmptyRepository(t *testing.T, ctx context.Context, provider BlobDescriptorCacheProvider) {
|
||||
if _, err := provider.Stat(ctx, "sha384:abc"); err != distribution.ErrBlobUnknown {
|
||||
t.Fatalf("expected unknown blob error with empty store: %v", err)
|
||||
}
|
||||
|
||||
cache, err := provider.RepositoryScoped("")
|
||||
if err == nil {
|
||||
t.Fatalf("expected error checking for cache item with empty repo")
|
||||
t.Fatalf("expected an error when asking for invalid repo")
|
||||
}
|
||||
|
||||
exists, err = lic.Contains(ctx, "foo/bar", "")
|
||||
if err == nil {
|
||||
t.Fatalf("expected error checking for cache item with empty digest")
|
||||
}
|
||||
|
||||
exists, err = lic.Contains(ctx, "foo/bar", "fake:abc")
|
||||
cache, err = provider.RepositoryScoped("foo/bar")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error checking for cache item: %v", err)
|
||||
t.Fatalf("unexpected error getting repository: %v", err)
|
||||
}
|
||||
|
||||
if exists {
|
||||
t.Fatalf("item should not exist")
|
||||
if err := cache.SetDescriptor(ctx, "", distribution.Descriptor{
|
||||
Digest: "sha384:abc",
|
||||
Length: 10,
|
||||
MediaType: "application/octet-stream"}); err != digest.ErrDigestInvalidFormat {
|
||||
t.Fatalf("expected error with invalid digest: %v", err)
|
||||
}
|
||||
|
||||
if err := lic.Add(ctx, "", "fake:abc"); err == nil {
|
||||
t.Fatalf("expected error adding cache item with empty name")
|
||||
if err := cache.SetDescriptor(ctx, "sha384:abc", distribution.Descriptor{
|
||||
Digest: "",
|
||||
Length: 10,
|
||||
MediaType: "application/octet-stream"}); err == nil {
|
||||
t.Fatalf("expected error setting value on invalid descriptor")
|
||||
}
|
||||
|
||||
if err := lic.Add(ctx, "foo/bar", ""); err == nil {
|
||||
t.Fatalf("expected error adding cache item with empty digest")
|
||||
if _, err := cache.Stat(ctx, ""); err != digest.ErrDigestInvalidFormat {
|
||||
t.Fatalf("expected error checking for cache item with empty digest: %v", err)
|
||||
}
|
||||
|
||||
if err := lic.Add(ctx, "foo/bar", "fake:abc"); err != nil {
|
||||
t.Fatalf("unexpected error adding item: %v", err)
|
||||
}
|
||||
|
||||
exists, err = lic.Contains(ctx, "foo/bar", "fake:abc")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error checking for cache item: %v", err)
|
||||
}
|
||||
|
||||
if !exists {
|
||||
t.Fatalf("item should exist")
|
||||
}
|
||||
|
||||
_, err = lic.Meta(ctx, "")
|
||||
if err == nil || err == ErrNotFound {
|
||||
t.Fatalf("expected error getting meta for cache item with empty digest")
|
||||
}
|
||||
|
||||
_, err = lic.Meta(ctx, "fake:abc")
|
||||
if err != ErrNotFound {
|
||||
t.Fatalf("expected unknown layer error getting meta for cache item with empty digest")
|
||||
}
|
||||
|
||||
if err = lic.SetMeta(ctx, "", LayerMeta{}); err == nil {
|
||||
t.Fatalf("expected error setting meta for cache item with empty digest")
|
||||
}
|
||||
|
||||
if err = lic.SetMeta(ctx, "foo/bar", LayerMeta{}); err == nil {
|
||||
t.Fatalf("expected error setting meta for cache item with empty meta")
|
||||
}
|
||||
|
||||
expected := LayerMeta{Path: "/foo/bar", Length: 20}
|
||||
if err := lic.SetMeta(ctx, "foo/bar", expected); err != nil {
|
||||
t.Fatalf("unexpected error setting meta: %v", err)
|
||||
}
|
||||
|
||||
meta, err := lic.Meta(ctx, "foo/bar")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error getting meta: %v", err)
|
||||
}
|
||||
|
||||
if meta != expected {
|
||||
t.Fatalf("retrieved meta data did not match: %v", err)
|
||||
if _, err := cache.Stat(ctx, "sha384:abc"); err != distribution.ErrBlobUnknown {
|
||||
t.Fatalf("expected unknown blob error with empty repo: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func checkBlobDescriptorCacheSetAndRead(t *testing.T, ctx context.Context, provider BlobDescriptorCacheProvider) {
|
||||
localDigest := digest.Digest("sha384:abc")
|
||||
expected := distribution.Descriptor{
|
||||
Digest: "sha256:abc",
|
||||
Length: 10,
|
||||
MediaType: "application/octet-stream"}
|
||||
|
||||
cache, err := provider.RepositoryScoped("foo/bar")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error getting scoped cache: %v", err)
|
||||
}
|
||||
|
||||
if err := cache.SetDescriptor(ctx, localDigest, expected); err != nil {
|
||||
t.Fatalf("error setting descriptor: %v", err)
|
||||
}
|
||||
|
||||
desc, err := cache.Stat(ctx, localDigest)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error statting fake2:abc: %v", err)
|
||||
}
|
||||
|
||||
if expected != desc {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", expected, desc)
|
||||
}
|
||||
|
||||
// also check that we set the canonical key ("fake:abc")
|
||||
desc, err = cache.Stat(ctx, localDigest)
|
||||
if err != nil {
|
||||
t.Fatalf("descriptor not returned for canonical key: %v", err)
|
||||
}
|
||||
|
||||
if expected != desc {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", expected, desc)
|
||||
}
|
||||
|
||||
// ensure that global gets extra descriptor mapping
|
||||
desc, err = provider.Stat(ctx, localDigest)
|
||||
if err != nil {
|
||||
t.Fatalf("expected blob unknown in global cache: %v, %v", err, desc)
|
||||
}
|
||||
|
||||
if desc != expected {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", expected, desc)
|
||||
}
|
||||
|
||||
// get at it through canonical descriptor
|
||||
desc, err = provider.Stat(ctx, expected.Digest)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error checking glboal descriptor: %v", err)
|
||||
}
|
||||
|
||||
if desc != expected {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", expected, desc)
|
||||
}
|
||||
|
||||
// now, we set the repo local mediatype to something else and ensure it
|
||||
// doesn't get changed in the provider cache.
|
||||
expected.MediaType = "application/json"
|
||||
|
||||
if err := cache.SetDescriptor(ctx, localDigest, expected); err != nil {
|
||||
t.Fatalf("unexpected error setting descriptor: %v", err)
|
||||
}
|
||||
|
||||
desc, err = cache.Stat(ctx, localDigest)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error getting descriptor: %v", err)
|
||||
}
|
||||
|
||||
if desc != expected {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", desc, expected)
|
||||
}
|
||||
|
||||
desc, err = provider.Stat(ctx, localDigest)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error getting global descriptor: %v", err)
|
||||
}
|
||||
|
||||
expected.MediaType = "application/octet-stream" // expect original mediatype in global
|
||||
|
||||
if desc != expected {
|
||||
t.Fatalf("unexpected descriptor: %#v != %#v", desc, expected)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
174
docs/storage/cache/memory.go
vendored
174
docs/storage/cache/memory.go
vendored
|
|
@ -1,63 +1,149 @@
|
|||
package cache
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/context"
|
||||
"github.com/docker/distribution/digest"
|
||||
"golang.org/x/net/context"
|
||||
"github.com/docker/distribution/registry/api/v2"
|
||||
)
|
||||
|
||||
// inmemoryLayerInfoCache is a map-based implementation of LayerInfoCache.
|
||||
type inmemoryLayerInfoCache struct {
|
||||
membership map[string]map[digest.Digest]struct{}
|
||||
meta map[digest.Digest]LayerMeta
|
||||
type inMemoryBlobDescriptorCacheProvider struct {
|
||||
global *mapBlobDescriptorCache
|
||||
repositories map[string]*mapBlobDescriptorCache
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewInMemoryLayerInfoCache provides an implementation of LayerInfoCache that
|
||||
// stores results in memory.
|
||||
func NewInMemoryLayerInfoCache() LayerInfoCache {
|
||||
return &base{&inmemoryLayerInfoCache{
|
||||
membership: make(map[string]map[digest.Digest]struct{}),
|
||||
meta: make(map[digest.Digest]LayerMeta),
|
||||
}}
|
||||
// NewInMemoryBlobDescriptorCacheProvider returns a new mapped-based cache for
|
||||
// storing blob descriptor data.
|
||||
func NewInMemoryBlobDescriptorCacheProvider() BlobDescriptorCacheProvider {
|
||||
return &inMemoryBlobDescriptorCacheProvider{
|
||||
global: newMapBlobDescriptorCache(),
|
||||
repositories: make(map[string]*mapBlobDescriptorCache),
|
||||
}
|
||||
}
|
||||
|
||||
func (ilic *inmemoryLayerInfoCache) Contains(ctx context.Context, repo string, dgst digest.Digest) (bool, error) {
|
||||
members, ok := ilic.membership[repo]
|
||||
if !ok {
|
||||
return false, nil
|
||||
func (imbdcp *inMemoryBlobDescriptorCacheProvider) RepositoryScoped(repo string) (distribution.BlobDescriptorService, error) {
|
||||
if err := v2.ValidateRespositoryName(repo); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
_, ok = members[dgst]
|
||||
return ok, nil
|
||||
imbdcp.mu.RLock()
|
||||
defer imbdcp.mu.RUnlock()
|
||||
|
||||
return &repositoryScopedInMemoryBlobDescriptorCache{
|
||||
repo: repo,
|
||||
parent: imbdcp,
|
||||
repository: imbdcp.repositories[repo],
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add adds the layer to the redis repository blob set.
|
||||
func (ilic *inmemoryLayerInfoCache) Add(ctx context.Context, repo string, dgst digest.Digest) error {
|
||||
members, ok := ilic.membership[repo]
|
||||
if !ok {
|
||||
members = make(map[digest.Digest]struct{})
|
||||
ilic.membership[repo] = members
|
||||
func (imbdcp *inMemoryBlobDescriptorCacheProvider) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
return imbdcp.global.Stat(ctx, dgst)
|
||||
}
|
||||
|
||||
func (imbdcp *inMemoryBlobDescriptorCacheProvider) SetDescriptor(ctx context.Context, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
_, err := imbdcp.Stat(ctx, dgst)
|
||||
if err == distribution.ErrBlobUnknown {
|
||||
|
||||
if dgst.Algorithm() != desc.Digest.Algorithm() && dgst != desc.Digest {
|
||||
// if the digests differ, set the other canonical mapping
|
||||
if err := imbdcp.global.SetDescriptor(ctx, desc.Digest, desc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// unknown, just set it
|
||||
return imbdcp.global.SetDescriptor(ctx, dgst, desc)
|
||||
}
|
||||
|
||||
members[dgst] = struct{}{}
|
||||
// we already know it, do nothing
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Meta retrieves the layer meta data from the redis hash, returning
|
||||
// ErrUnknownLayer if not found.
|
||||
func (ilic *inmemoryLayerInfoCache) Meta(ctx context.Context, dgst digest.Digest) (LayerMeta, error) {
|
||||
meta, ok := ilic.meta[dgst]
|
||||
if !ok {
|
||||
return LayerMeta{}, ErrNotFound
|
||||
}
|
||||
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// SetMeta sets the meta data for the given digest using a redis hash. A hash
|
||||
// is used here since we may store unrelated fields about a layer in the
|
||||
// future.
|
||||
func (ilic *inmemoryLayerInfoCache) SetMeta(ctx context.Context, dgst digest.Digest, meta LayerMeta) error {
|
||||
ilic.meta[dgst] = meta
|
||||
// repositoryScopedInMemoryBlobDescriptorCache provides the request scoped
|
||||
// repository cache. Instances are not thread-safe but the delegated
|
||||
// operations are.
|
||||
type repositoryScopedInMemoryBlobDescriptorCache struct {
|
||||
repo string
|
||||
parent *inMemoryBlobDescriptorCacheProvider // allows lazy allocation of repo's map
|
||||
repository *mapBlobDescriptorCache
|
||||
}
|
||||
|
||||
func (rsimbdcp *repositoryScopedInMemoryBlobDescriptorCache) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
if rsimbdcp.repository == nil {
|
||||
return distribution.Descriptor{}, distribution.ErrBlobUnknown
|
||||
}
|
||||
|
||||
return rsimbdcp.repository.Stat(ctx, dgst)
|
||||
}
|
||||
|
||||
func (rsimbdcp *repositoryScopedInMemoryBlobDescriptorCache) SetDescriptor(ctx context.Context, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if rsimbdcp.repository == nil {
|
||||
// allocate map since we are setting it now.
|
||||
rsimbdcp.parent.mu.Lock()
|
||||
var ok bool
|
||||
// have to read back value since we may have allocated elsewhere.
|
||||
rsimbdcp.repository, ok = rsimbdcp.parent.repositories[rsimbdcp.repo]
|
||||
if !ok {
|
||||
rsimbdcp.repository = newMapBlobDescriptorCache()
|
||||
rsimbdcp.parent.repositories[rsimbdcp.repo] = rsimbdcp.repository
|
||||
}
|
||||
|
||||
rsimbdcp.parent.mu.Unlock()
|
||||
}
|
||||
|
||||
if err := rsimbdcp.repository.SetDescriptor(ctx, dgst, desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return rsimbdcp.parent.SetDescriptor(ctx, dgst, desc)
|
||||
}
|
||||
|
||||
// mapBlobDescriptorCache provides a simple map-based implementation of the
|
||||
// descriptor cache.
|
||||
type mapBlobDescriptorCache struct {
|
||||
descriptors map[digest.Digest]distribution.Descriptor
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
var _ distribution.BlobDescriptorService = &mapBlobDescriptorCache{}
|
||||
|
||||
func newMapBlobDescriptorCache() *mapBlobDescriptorCache {
|
||||
return &mapBlobDescriptorCache{
|
||||
descriptors: make(map[digest.Digest]distribution.Descriptor),
|
||||
}
|
||||
}
|
||||
|
||||
func (mbdc *mapBlobDescriptorCache) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
mbdc.mu.RLock()
|
||||
defer mbdc.mu.RUnlock()
|
||||
|
||||
desc, ok := mbdc.descriptors[dgst]
|
||||
if !ok {
|
||||
return distribution.Descriptor{}, distribution.ErrBlobUnknown
|
||||
}
|
||||
|
||||
return desc, nil
|
||||
}
|
||||
|
||||
func (mbdc *mapBlobDescriptorCache) SetDescriptor(ctx context.Context, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := validateDescriptor(desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
mbdc.mu.Lock()
|
||||
defer mbdc.mu.Unlock()
|
||||
|
||||
mbdc.descriptors[dgst] = desc
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
6
docs/storage/cache/memory_test.go
vendored
6
docs/storage/cache/memory_test.go
vendored
|
|
@ -2,8 +2,8 @@ package cache
|
|||
|
||||
import "testing"
|
||||
|
||||
// TestInMemoryLayerInfoCache checks the in memory implementation is working
|
||||
// TestInMemoryBlobInfoCache checks the in memory implementation is working
|
||||
// correctly.
|
||||
func TestInMemoryLayerInfoCache(t *testing.T) {
|
||||
checkLayerInfoCache(t, NewInMemoryLayerInfoCache())
|
||||
func TestInMemoryBlobInfoCache(t *testing.T) {
|
||||
checkBlobDescriptorCache(t, NewInMemoryBlobDescriptorCacheProvider())
|
||||
}
|
||||
|
|
|
|||
238
docs/storage/cache/redis.go
vendored
238
docs/storage/cache/redis.go
vendored
|
|
@ -1,20 +1,28 @@
|
|||
package cache
|
||||
|
||||
import (
|
||||
ctxu "github.com/docker/distribution/context"
|
||||
"fmt"
|
||||
|
||||
"github.com/docker/distribution/registry/api/v2"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/context"
|
||||
"github.com/docker/distribution/digest"
|
||||
"github.com/garyburd/redigo/redis"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
||||
// redisLayerInfoCache provides an implementation of storage.LayerInfoCache
|
||||
// based on redis. Layer info is stored in two parts. The first provide fast
|
||||
// access to repository membership through a redis set for each repo. The
|
||||
// second is a redis hash keyed by the digest of the layer, providing path and
|
||||
// length information. Note that there is no implied relationship between
|
||||
// these two caches. The layer may exist in one, both or none and the code
|
||||
// must be written this way.
|
||||
type redisLayerInfoCache struct {
|
||||
// redisBlobStatService provides an implementation of
|
||||
// BlobDescriptorCacheProvider based on redis. Blob descritors are stored in
|
||||
// two parts. The first provide fast access to repository membership through a
|
||||
// redis set for each repo. The second is a redis hash keyed by the digest of
|
||||
// the layer, providing path, length and mediatype information. There is also
|
||||
// a per-repository redis hash of the blob descriptor, allowing override of
|
||||
// data. This is currently used to override the mediatype on a per-repository
|
||||
// basis.
|
||||
//
|
||||
// Note that there is no implied relationship between these two caches. The
|
||||
// layer may exist in one, both or none and the code must be written this way.
|
||||
type redisBlobDescriptorService struct {
|
||||
pool *redis.Pool
|
||||
|
||||
// TODO(stevvooe): We use a pool because we don't have great control over
|
||||
|
|
@ -23,76 +31,194 @@ type redisLayerInfoCache struct {
|
|||
// request objects, we can change this to a connection.
|
||||
}
|
||||
|
||||
// NewRedisLayerInfoCache returns a new redis-based LayerInfoCache using the
|
||||
// provided redis connection pool.
|
||||
func NewRedisLayerInfoCache(pool *redis.Pool) LayerInfoCache {
|
||||
return &base{&redisLayerInfoCache{
|
||||
var _ BlobDescriptorCacheProvider = &redisBlobDescriptorService{}
|
||||
|
||||
// NewRedisBlobDescriptorCacheProvider returns a new redis-based
|
||||
// BlobDescriptorCacheProvider using the provided redis connection pool.
|
||||
func NewRedisBlobDescriptorCacheProvider(pool *redis.Pool) BlobDescriptorCacheProvider {
|
||||
return &redisBlobDescriptorService{
|
||||
pool: pool,
|
||||
}}
|
||||
}
|
||||
}
|
||||
|
||||
// Contains does a membership check on the repository blob set in redis. This
|
||||
// is used as an access check before looking up global path information. If
|
||||
// false is returned, the caller should still check the backend to if it
|
||||
// exists elsewhere.
|
||||
func (rlic *redisLayerInfoCache) Contains(ctx context.Context, repo string, dgst digest.Digest) (bool, error) {
|
||||
conn := rlic.pool.Get()
|
||||
defer conn.Close()
|
||||
// RepositoryScoped returns the scoped cache.
|
||||
func (rbds *redisBlobDescriptorService) RepositoryScoped(repo string) (distribution.BlobDescriptorService, error) {
|
||||
if err := v2.ValidateRespositoryName(repo); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ctxu.GetLogger(ctx).Debugf("(*redisLayerInfoCache).Contains(%q, %q)", repo, dgst)
|
||||
return redis.Bool(conn.Do("SISMEMBER", rlic.repositoryBlobSetKey(repo), dgst))
|
||||
return &repositoryScopedRedisBlobDescriptorService{
|
||||
repo: repo,
|
||||
upstream: rbds,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add adds the layer to the redis repository blob set.
|
||||
func (rlic *redisLayerInfoCache) Add(ctx context.Context, repo string, dgst digest.Digest) error {
|
||||
conn := rlic.pool.Get()
|
||||
// Stat retrieves the descriptor data from the redis hash entry.
|
||||
func (rbds *redisBlobDescriptorService) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
conn := rbds.pool.Get()
|
||||
defer conn.Close()
|
||||
|
||||
ctxu.GetLogger(ctx).Debugf("(*redisLayerInfoCache).Add(%q, %q)", repo, dgst)
|
||||
_, err := conn.Do("SADD", rlic.repositoryBlobSetKey(repo), dgst)
|
||||
return err
|
||||
return rbds.stat(ctx, conn, dgst)
|
||||
}
|
||||
|
||||
// Meta retrieves the layer meta data from the redis hash, returning
|
||||
// ErrUnknownLayer if not found.
|
||||
func (rlic *redisLayerInfoCache) Meta(ctx context.Context, dgst digest.Digest) (LayerMeta, error) {
|
||||
conn := rlic.pool.Get()
|
||||
defer conn.Close()
|
||||
|
||||
reply, err := redis.Values(conn.Do("HMGET", rlic.blobMetaHashKey(dgst), "path", "length"))
|
||||
// stat provides an internal stat call that takes a connection parameter. This
|
||||
// allows some internal management of the connection scope.
|
||||
func (rbds *redisBlobDescriptorService) stat(ctx context.Context, conn redis.Conn, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
reply, err := redis.Values(conn.Do("HMGET", rbds.blobDescriptorHashKey(dgst), "digest", "length", "mediatype"))
|
||||
if err != nil {
|
||||
return LayerMeta{}, err
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
if len(reply) < 2 || reply[0] == nil || reply[1] == nil {
|
||||
return LayerMeta{}, ErrNotFound
|
||||
if len(reply) < 2 || reply[0] == nil || reply[1] == nil { // don't care if mediatype is nil
|
||||
return distribution.Descriptor{}, distribution.ErrBlobUnknown
|
||||
}
|
||||
|
||||
var meta LayerMeta
|
||||
if _, err := redis.Scan(reply, &meta.Path, &meta.Length); err != nil {
|
||||
return LayerMeta{}, err
|
||||
var desc distribution.Descriptor
|
||||
if _, err := redis.Scan(reply, &desc.Digest, &desc.Length, &desc.MediaType); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
return meta, nil
|
||||
return desc, nil
|
||||
}
|
||||
|
||||
// SetMeta sets the meta data for the given digest using a redis hash. A hash
|
||||
// is used here since we may store unrelated fields about a layer in the
|
||||
// future.
|
||||
func (rlic *redisLayerInfoCache) SetMeta(ctx context.Context, dgst digest.Digest, meta LayerMeta) error {
|
||||
conn := rlic.pool.Get()
|
||||
// SetDescriptor sets the descriptor data for the given digest using a redis
|
||||
// hash. A hash is used here since we may store unrelated fields about a layer
|
||||
// in the future.
|
||||
func (rbds *redisBlobDescriptorService) SetDescriptor(ctx context.Context, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := validateDescriptor(desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
conn := rbds.pool.Get()
|
||||
defer conn.Close()
|
||||
|
||||
_, err := conn.Do("HMSET", rlic.blobMetaHashKey(dgst), "path", meta.Path, "length", meta.Length)
|
||||
return err
|
||||
return rbds.setDescriptor(ctx, conn, dgst, desc)
|
||||
}
|
||||
|
||||
// repositoryBlobSetKey returns the key for the blob set in the cache.
|
||||
func (rlic *redisLayerInfoCache) repositoryBlobSetKey(repo string) string {
|
||||
return "repository::" + repo + "::blobs"
|
||||
func (rbds *redisBlobDescriptorService) setDescriptor(ctx context.Context, conn redis.Conn, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if _, err := conn.Do("HMSET", rbds.blobDescriptorHashKey(dgst),
|
||||
"digest", desc.Digest,
|
||||
"length", desc.Length); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Only set mediatype if not already set.
|
||||
if _, err := conn.Do("HSETNX", rbds.blobDescriptorHashKey(dgst),
|
||||
"mediatype", desc.MediaType); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// blobMetaHashKey returns the cache key for immutable blob meta data.
|
||||
func (rlic *redisLayerInfoCache) blobMetaHashKey(dgst digest.Digest) string {
|
||||
func (rbds *redisBlobDescriptorService) blobDescriptorHashKey(dgst digest.Digest) string {
|
||||
return "blobs::" + dgst.String()
|
||||
}
|
||||
|
||||
type repositoryScopedRedisBlobDescriptorService struct {
|
||||
repo string
|
||||
upstream *redisBlobDescriptorService
|
||||
}
|
||||
|
||||
var _ distribution.BlobDescriptorService = &repositoryScopedRedisBlobDescriptorService{}
|
||||
|
||||
// Stat ensures that the digest is a member of the specified repository and
|
||||
// forwards the descriptor request to the global blob store. If the media type
|
||||
// differs for the repository, we override it.
|
||||
func (rsrbds *repositoryScopedRedisBlobDescriptorService) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
conn := rsrbds.upstream.pool.Get()
|
||||
defer conn.Close()
|
||||
|
||||
// Check membership to repository first
|
||||
member, err := redis.Bool(conn.Do("SISMEMBER", rsrbds.repositoryBlobSetKey(rsrbds.repo), dgst))
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
if !member {
|
||||
return distribution.Descriptor{}, distribution.ErrBlobUnknown
|
||||
}
|
||||
|
||||
upstream, err := rsrbds.upstream.stat(ctx, conn, dgst)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
// We allow a per repository mediatype, let's look it up here.
|
||||
mediatype, err := redis.String(conn.Do("HGET", rsrbds.blobDescriptorHashKey(dgst), "mediatype"))
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
if mediatype != "" {
|
||||
upstream.MediaType = mediatype
|
||||
}
|
||||
|
||||
return upstream, nil
|
||||
}
|
||||
|
||||
func (rsrbds *repositoryScopedRedisBlobDescriptorService) SetDescriptor(ctx context.Context, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if err := validateDigest(dgst); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := validateDescriptor(desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if dgst != desc.Digest {
|
||||
if dgst.Algorithm() == desc.Digest.Algorithm() {
|
||||
return fmt.Errorf("redis cache: digest for descriptors differ but algorthim does not: %q != %q", dgst, desc.Digest)
|
||||
}
|
||||
}
|
||||
|
||||
conn := rsrbds.upstream.pool.Get()
|
||||
defer conn.Close()
|
||||
|
||||
return rsrbds.setDescriptor(ctx, conn, dgst, desc)
|
||||
}
|
||||
|
||||
func (rsrbds *repositoryScopedRedisBlobDescriptorService) setDescriptor(ctx context.Context, conn redis.Conn, dgst digest.Digest, desc distribution.Descriptor) error {
|
||||
if _, err := conn.Do("SADD", rsrbds.repositoryBlobSetKey(rsrbds.repo), dgst); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := rsrbds.upstream.setDescriptor(ctx, conn, dgst, desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Override repository mediatype.
|
||||
if _, err := conn.Do("HSET", rsrbds.blobDescriptorHashKey(dgst), "mediatype", desc.MediaType); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Also set the values for the primary descriptor, if they differ by
|
||||
// algorithm (ie sha256 vs tarsum).
|
||||
if desc.Digest != "" && dgst != desc.Digest && dgst.Algorithm() != desc.Digest.Algorithm() {
|
||||
if err := rsrbds.setDescriptor(ctx, conn, desc.Digest, desc); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rsrbds *repositoryScopedRedisBlobDescriptorService) blobDescriptorHashKey(dgst digest.Digest) string {
|
||||
return "repository::" + rsrbds.repo + "::blobs::" + dgst.String()
|
||||
}
|
||||
|
||||
func (rsrbds *repositoryScopedRedisBlobDescriptorService) repositoryBlobSetKey(repo string) string {
|
||||
return "repository::" + rsrbds.repo + "::blobs"
|
||||
}
|
||||
|
|
|
|||
4
docs/storage/cache/redis_test.go
vendored
4
docs/storage/cache/redis_test.go
vendored
|
|
@ -17,7 +17,7 @@ func init() {
|
|||
|
||||
// TestRedisLayerInfoCache exercises a live redis instance using the cache
|
||||
// implementation.
|
||||
func TestRedisLayerInfoCache(t *testing.T) {
|
||||
func TestRedisBlobDescriptorCacheProvider(t *testing.T) {
|
||||
if redisAddr == "" {
|
||||
// fallback to an environement variable
|
||||
redisAddr = os.Getenv("TEST_REGISTRY_STORAGE_CACHE_REDIS_ADDR")
|
||||
|
|
@ -46,5 +46,5 @@ func TestRedisLayerInfoCache(t *testing.T) {
|
|||
t.Fatalf("unexpected error flushing redis db: %v", err)
|
||||
}
|
||||
|
||||
checkLayerInfoCache(t, NewRedisLayerInfoCache(pool))
|
||||
checkBlobDescriptorCache(t, NewRedisBlobDescriptorCacheProvider(pool))
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue