Merge pull request #740 from stevvooe/disable-storage-redirects

Allow disabling of storage driver redirects

configuration/configuration.go

@@ -242,6 +242,8 @@ func (storage Storage) Type() string {
 			// allow configuration of caching
 		case "delete":
 			// allow configuration of delete
+		case "redirect":
+			// allow configuration of redirect
 		default:
 			return k
 		}
@@ -275,7 +277,8 @@ func (storage *Storage) UnmarshalYAML(unmarshal func(interface{}) error) error {
 					// allow configuration of caching
 				case "delete":
 					// allow configuration of delete
-
+				case "redirect":
+					// allow configuration of redirect
 				default:
 					types = append(types, k)
 				}

docs/configuration.md

@@ -104,6 +104,8 @@ information about each option that appears later in this page.
         region: fr
         container: containername
         rootdirectory: /swift/object/name/prefix
+      redirect:
+        disable: false
       cache:
         blobdescriptor: redis
       maintenance:
@@ -328,6 +330,8 @@ Permitted values are `error`, `warn`, `info` and `debug`. The default is
           age: 168h
           interval: 24h
           dryrun: false
+      redirect:
+        disable: false
 
 The storage option is **required** and defines which storage backend is in use.
 You must configure one backend; if you configure more, the registry returns an error.
@@ -350,6 +354,21 @@ map.
 >are equivalent, `layerinfo` has been deprecated, in favor or
 >`blobdescriptor`.
 
+### redirect
+
+The `redirect` subsection provides configuration for managing redirects from
+content backends. For backends that support it, redirecting is enabled by
+default. Certain deployment scenarios may prefer to route all data through the
+Registry, rather than redirecting to the backend. This may be more efficient
+when using a backend that is not colocated or when a registry instance is
+doing aggressive caching.
+
+Redirects can be disabled by adding a single flag `disable`, set to `true`
+under the `redirect` section:
+
+  redirect:
+    disable: true
+
 ### filesystem
 
 The `filesystem` storage backend uses the local disk to store registry files. It

notifications/listener_test.go

@@ -18,7 +18,7 @@ import (
 
 func TestListener(t *testing.T) {
 	ctx := context.Background()
-	registry := storage.NewRegistryWithDriver(ctx, inmemory.New(), memory.NewInMemoryBlobDescriptorCacheProvider(), true)
+	registry := storage.NewRegistryWithDriver(ctx, inmemory.New(), memory.NewInMemoryBlobDescriptorCacheProvider(), true, true)
 	tl := &testListener{
 		ops: make(map[string]int),
 	}

registry/handlers/app.go

@@ -112,7 +112,8 @@ func NewApp(ctx context.Context, configuration configuration.Configuration) *App
 	app.configureRedis(&configuration)
 	app.configureLogHook(&configuration)
 
-	deleteEnabled := false
+	// configure deletion
+	var deleteEnabled bool
 	if d, ok := configuration.Storage["delete"]; ok {
 		e, ok := d["enabled"]
 		if ok {
@@ -122,6 +123,22 @@ func NewApp(ctx context.Context, configuration configuration.Configuration) *App
 		}
 	}
 
+	// configure redirects
+	var redirectDisabled bool
+	if redirectConfig, ok := configuration.Storage["redirect"]; ok {
+		v := redirectConfig["disable"]
+		switch v := v.(type) {
+		case bool:
+			redirectDisabled = v
+		default:
+			panic(fmt.Sprintf("invalid type for redirect config: %#v", redirectConfig))
+		}
+
+		if redirectDisabled {
+			ctxu.GetLogger(app).Infof("backend redirection disabled")
+		}
+	}
+
 	// configure storage caches
 	if cc, ok := configuration.Storage["cache"]; ok {
 		v, ok := cc["blobdescriptor"]
@@ -135,10 +152,10 @@ func NewApp(ctx context.Context, configuration configuration.Configuration) *App
 			if app.redis == nil {
 				panic("redis configuration required to use for layerinfo cache")
 			}
-			app.registry = storage.NewRegistryWithDriver(app, app.driver, rediscache.NewRedisBlobDescriptorCacheProvider(app.redis), deleteEnabled)
+			app.registry = storage.NewRegistryWithDriver(app, app.driver, rediscache.NewRedisBlobDescriptorCacheProvider(app.redis), deleteEnabled, !redirectDisabled)
 			ctxu.GetLogger(app).Infof("using redis blob descriptor cache")
 		case "inmemory":
-			app.registry = storage.NewRegistryWithDriver(app, app.driver, memorycache.NewInMemoryBlobDescriptorCacheProvider(), deleteEnabled)
+			app.registry = storage.NewRegistryWithDriver(app, app.driver, memorycache.NewInMemoryBlobDescriptorCacheProvider(), deleteEnabled, !redirectDisabled)
 			ctxu.GetLogger(app).Infof("using inmemory blob descriptor cache")
 		default:
 			if v != "" {
@@ -149,7 +166,7 @@ func NewApp(ctx context.Context, configuration configuration.Configuration) *App
 
 	if app.registry == nil {
 		// configure the registry if no cache section is available.
-		app.registry = storage.NewRegistryWithDriver(app.Context, app.driver, nil, deleteEnabled)
+		app.registry = storage.NewRegistryWithDriver(app.Context, app.driver, nil, deleteEnabled, !redirectDisabled)
 	}
 
 	app.registry, err = applyRegistryMiddleware(app.registry, configuration.Middleware["registry"])

registry/handlers/app_test.go

@@ -31,7 +31,7 @@ func TestAppDispatcher(t *testing.T) {
 		Context:  ctx,
 		router:   v2.Router(),
 		driver:   driver,
-		registry: storage.NewRegistryWithDriver(ctx, driver, memorycache.NewInMemoryBlobDescriptorCacheProvider(), true),
+		registry: storage.NewRegistryWithDriver(ctx, driver, memorycache.NewInMemoryBlobDescriptorCacheProvider(), true, true),
 	}
 	server := httptest.NewServer(app)
 	router := v2.Router()

registry/storage/blob_test.go

@@ -33,7 +33,7 @@ func TestSimpleBlobUpload(t *testing.T) {
 	ctx := context.Background()
 	imageName := "foo/bar"
 	driver := inmemory.New()
-	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true)
+	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true, true)
 	repository, err := registry.Repository(ctx, imageName)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)
@@ -193,7 +193,7 @@ func TestSimpleBlobUpload(t *testing.T) {
 	}
 
 	// Reuse state to test delete with a delete-disabled registry
-	registry = NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), false)
+	registry = NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), false, true)
 	repository, err = registry.Repository(ctx, imageName)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)
@@ -212,7 +212,7 @@ func TestSimpleBlobRead(t *testing.T) {
 	ctx := context.Background()
 	imageName := "foo/bar"
 	driver := inmemory.New()
-	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true)
+	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true, true)
 	repository, err := registry.Repository(ctx, imageName)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)
@@ -316,7 +316,7 @@ func TestLayerUploadZeroLength(t *testing.T) {
 	ctx := context.Background()
 	imageName := "foo/bar"
 	driver := inmemory.New()
-	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true)
+	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true, true)
 	repository, err := registry.Repository(ctx, imageName)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)

registry/storage/blobserver.go

@@ -20,6 +20,7 @@ type blobServer struct {
 	driver   driver.StorageDriver
 	statter  distribution.BlobStatter
 	pathFn   func(dgst digest.Digest) (string, error)
+	redirect bool // allows disabling URLFor redirects
 }
 
 func (bs *blobServer) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
@@ -37,8 +38,13 @@ func (bs *blobServer) ServeBlob(ctx context.Context, w http.ResponseWriter, r *h
 
 	switch err {
 	case nil:
+		if bs.redirect {
 			// Redirect to storage URL.
 			http.Redirect(w, r, redirectURL, http.StatusTemporaryRedirect)
+			return err
+		}
+
+		fallthrough
 	case driver.ErrUnsupportedMethod:
 		// Fallback to serving the content directly.
 		br, err := newFileReader(ctx, bs.driver, path, desc.Size)

registry/storage/catalog_test.go

@@ -22,7 +22,7 @@ func setupFS(t *testing.T) *setupEnv {
 	d := inmemory.New()
 	c := []byte("")
 	ctx := context.Background()
-	registry := NewRegistryWithDriver(ctx, d, memory.NewInMemoryBlobDescriptorCacheProvider(), false)
+	registry := NewRegistryWithDriver(ctx, d, memory.NewInMemoryBlobDescriptorCacheProvider(), false, true)
 	rootpath, _ := defaultPathMapper.path(repositoriesRootPathSpec{})
 
 	repos := []string{

registry/storage/manifeststore_test.go

@@ -29,7 +29,8 @@ type manifestStoreTestEnv struct {
 func newManifestStoreTestEnv(t *testing.T, name, tag string) *manifestStoreTestEnv {
 	ctx := context.Background()
 	driver := inmemory.New()
-	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true)
+	registry := NewRegistryWithDriver(ctx, driver, memory.NewInMemoryBlobDescriptorCacheProvider(), true, true)
+
 	repo, err := registry.Repository(ctx, name)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)
@@ -347,7 +348,7 @@ func TestManifestStorage(t *testing.T) {
 		t.Errorf("Deleted manifest get returned non-nil")
 	}
 
-	r := NewRegistryWithDriver(ctx, env.driver, memory.NewInMemoryBlobDescriptorCacheProvider(), false)
+	r := NewRegistryWithDriver(ctx, env.driver, memory.NewInMemoryBlobDescriptorCacheProvider(), false, true)
 	repo, err := r.Repository(ctx, env.name)
 	if err != nil {
 		t.Fatalf("unexpected error getting repo: %v", err)

registry/storage/registry.go

@@ -20,9 +20,12 @@ type registry struct {
 
 // NewRegistryWithDriver creates a new registry instance from the provided
 // driver. The resulting registry may be shared by multiple goroutines but is
-// cheap to allocate.
+// cheap to allocate. If redirect is true, the backend blob server will
-func NewRegistryWithDriver(ctx context.Context, driver storagedriver.StorageDriver, blobDescriptorCacheProvider cache.BlobDescriptorCacheProvider, deleteEnabled bool) distribution.Namespace {
+// attempt to use (StorageDriver).URLFor to serve all blobs.
-
+//
+// TODO(stevvooe): This function signature is getting out of hand. Move to
+// functional options for instance configuration.
+func NewRegistryWithDriver(ctx context.Context, driver storagedriver.StorageDriver, blobDescriptorCacheProvider cache.BlobDescriptorCacheProvider, deleteEnabled bool, redirect bool) distribution.Namespace {
 	// create global statter, with cache.
 	var statter distribution.BlobDescriptorService = &blobStatter{
 		driver: driver,
@@ -45,6 +48,7 @@ func NewRegistryWithDriver(ctx context.Context, driver storagedriver.StorageDriv
 			driver:   driver,
 			statter:  statter,
 			pathFn:   bs.path,
+			redirect: redirect,
 		},
 		blobDescriptorCacheProvider: blobDescriptorCacheProvider,
 		deleteEnabled:               deleteEnabled,