Move hacks for middlewares into a separate package

Author: Oleg Bulatov

pkg/dockerregistry/server/app.go

@@ -6,29 +6,33 @@ import (
 	"github.com/docker/distribution"
 	"github.com/docker/distribution/configuration"
 	"github.com/docker/distribution/context"
-	"github.com/docker/distribution/registry/handlers"
 	storagedriver "github.com/docker/distribution/registry/storage/driver"
 
 	"github.com/openshift/image-registry/pkg/dockerregistry/server/client"
 	registryconfig "github.com/openshift/image-registry/pkg/dockerregistry/server/configuration"
 	"github.com/openshift/image-registry/pkg/dockerregistry/server/maxconnections"
+	"github.com/openshift/image-registry/pkg/dockerregistry/server/supermiddleware"
 )
 
 const (
 	// Default values
 	defaultDigestToRepositoryCacheSize = 2048
 )
 
+// appMiddleware should be used only in tests.
+type appMiddleware interface {
+	Apply(supermiddleware.App) supermiddleware.App
+}
+
 // App is a global registry application object. Shared resources can be placed
 // on this object that will be accessible from all requests.
 type App struct {
 	// ctx is the parent context.
 	ctx context.Context
 
-	registryClient   client.RegistryClient
-	extraConfig      *registryconfig.Configuration
-	repositoryConfig repositoryConfig
-	writeLimiter     maxconnections.Limiter
+	registryClient client.RegistryClient
+	extraConfig    *registryconfig.Configuration
+	writeLimiter   maxconnections.Limiter
 
 	// driver gives access to the blob store.
 	// This variable holds the object created by docker/distribution. We
@@ -57,6 +61,16 @@ type App struct {
 	quotaEnforcing *quotaEnforcingConfig
 }
 
+func (app *App) Storage(driver storagedriver.StorageDriver, options map[string]interface{}) (storagedriver.StorageDriver, error) {
+	app.driver = driver
+	return driver, nil
+}
+
+func (app *App) Registry(registry distribution.Namespace, options map[string]interface{}) (distribution.Namespace, error) {
+	app.registry = registry
+	return registry, nil
+}
+
 // NewApp configures the registry application and returns http.Handler for it.
 // The program will be terminated if an error happens.
 func NewApp(ctx context.Context, registryClient client.RegistryClient, dockerConfig *configuration.Configuration, extraConfig *registryconfig.Configuration, writeLimiter maxconnections.Limiter) http.Handler {
@@ -74,35 +88,21 @@ func NewApp(ctx context.Context, registryClient client.RegistryClient, dockerCon
 	}
 	app.cachedLayers = cache
 
-	weaveAppIntoConfig(app, dockerConfig)
-
-	repositoryEnabled := false
-	for _, middleware := range dockerConfig.Middleware["repository"] {
-		if middleware.Name == middlewareOpenShift {
-			rc, err := newRepositoryConfig(ctx, extraConfig, middleware.Options)
-			if err != nil {
-				context.GetLogger(ctx).Fatalf("error configuring the repository middleware: %s", err)
-			}
-			app.repositoryConfig = rc
-			repositoryEnabled = true
-			break
-		}
+	superapp := supermiddleware.App(app)
+	if am := appMiddlewareFrom(ctx); am != nil {
+		superapp = am.Apply(superapp)
 	}
+	dockerApp := supermiddleware.NewApp(ctx, dockerConfig, superapp)
 
-	dockerApp := handlers.NewApp(ctx, dockerConfig)
-
-	if repositoryEnabled {
-		if app.driver == nil {
-			context.GetLogger(ctx).Fatalf("configuration error: the storage driver middleware %q is not activated", middlewareOpenShift)
-		}
-
-		if app.registry == nil {
-			context.GetLogger(ctx).Fatalf("configuration error: the registry middleware %q is not activated", middlewareOpenShift)
-		}
+	if app.driver == nil {
+		context.GetLogger(ctx).Fatalf("configuration error: the storage driver middleware %q is not activated", supermiddleware.Name)
+	}
+	if app.registry == nil {
+		context.GetLogger(ctx).Fatalf("configuration error: the registry middleware %q is not activated", supermiddleware.Name)
 	}
 
 	// Add a token handling endpoint
-	if dockerConfig.Auth.Type() == middlewareOpenShift {
+	if dockerConfig.Auth.Type() == supermiddleware.Name {
 		tokenRealm, err := registryconfig.TokenRealm(extraConfig.Auth.TokenRealm)
 		if err != nil {
 			context.GetLogger(dockerApp).Fatalf("error setting up token auth: %s", err)

pkg/dockerregistry/server/auth.go

@@ -89,7 +89,8 @@ var (
 	ErrUnsupportedResource = errors.New("unsupported resource")
 )
 
-func (app *App) newAccessController(authcfg *configuration.Auth) (registryauth.AccessController, error) {
+func (app *App) Auth(options map[string]interface{}) (registryauth.AccessController, error) {
+	authcfg := app.extraConfig.Auth
 	tokenRealm, err := configuration.TokenRealm(authcfg.TokenRealm)
 	if err != nil {
 		return nil, err

pkg/dockerregistry/server/auth_test.go

@@ -433,7 +433,7 @@ func TestAccessController(t *testing.T) {
 			if err := configuration.InitExtraConfig(config, app.extraConfig); err != nil {
 				t.Fatal(err)
 			}
-			accessController, err := app.newAccessController(app.extraConfig.Auth)
+			accessController, err := app.Auth(nil)
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/dockerregistry/server/blobdescriptorservice.go

@@ -1,17 +1,13 @@
 package server
 
 import (
-	"fmt"
 	"sort"
 	"time"
 
-	"github.com/Sirupsen/logrus"
 	"github.com/docker/distribution"
 	"github.com/docker/distribution/context"
 	"github.com/docker/distribution/digest"
 	"github.com/docker/distribution/manifest/schema2"
-	"github.com/docker/distribution/registry/middleware/registry"
-	"github.com/docker/distribution/registry/storage"
 
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 
@@ -34,61 +30,50 @@ func (b ByGeneration) Less(i, j int) bool { return b[i].Generation > b[j].Genera
 func (b ByGeneration) Len() int           { return len(b) }
 func (b ByGeneration) Swap(i, j int)      { b[i], b[j] = b[j], b[i] }
 
-func init() {
-	err := middleware.RegisterOptions(storage.BlobDescriptorServiceFactory(&blobDescriptorServiceFactory{}))
-	if err != nil {
-		logrus.Fatalf("Unable to register BlobDescriptorServiceFactory: %v", err)
-	}
-}
+type blobDescriptorServiceFactoryFunc func(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService
 
-// blobDescriptorServiceFactory needs to be able to work with blobs
-// directly without using links. This allows us to ignore the distribution
-// of blobs between repositories.
-type blobDescriptorServiceFactory struct{}
-
-func (bf *blobDescriptorServiceFactory) BlobAccessController(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService {
-	return &blobDescriptorService{svc}
+func (f blobDescriptorServiceFactoryFunc) BlobAccessController(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService {
+	return f(svc)
 }
 
 type blobDescriptorService struct {
 	distribution.BlobDescriptorService
+	repo *repository
+}
+
+func (r *repository) BlobDescriptorService(svc distribution.BlobDescriptorService) distribution.BlobDescriptorService {
+	return &blobDescriptorService{svc, r}
 }
 
 // Stat returns a a blob descriptor if the given blob is either linked in repository or is referenced in
 // corresponding image stream. This method is invoked from inside of upstream's linkedBlobStore. It expects
 // a proper repository object to be set on given context by upper openshift middleware wrappers.
 func (bs *blobDescriptorService) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
 	context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: starting with digest=%s", dgst.String())
-	repo, found := repositoryFrom(ctx)
-	if !found || repo == nil {
-		err := fmt.Errorf("failed to retrieve repository from context")
-		context.GetLogger(ctx).Error(err)
-		return distribution.Descriptor{}, err
-	}
 
 	// if there is a repo layer link, return its descriptor
 	desc, err := bs.BlobDescriptorService.Stat(ctx, dgst)
 	if err == nil {
 		// and remember the association
-		repo.cachedLayers.RememberDigest(dgst, repo.config.blobRepositoryCacheTTL, imageapi.DockerImageReference{
-			Namespace: repo.namespace,
-			Name:      repo.name,
+		bs.repo.cachedLayers.RememberDigest(dgst, bs.repo.app.extraConfig.Cache.BlobRepositoryTTL, imageapi.DockerImageReference{
+			Namespace: bs.repo.namespace,
+			Name:      bs.repo.name,
 		}.Exact())
 		return desc, nil
 	}
 
-	context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: could not stat layer link %s in repository %s: %v", dgst.String(), repo.Named().Name(), err)
+	context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: could not stat layer link %s in repository %s: %v", dgst.String(), bs.repo.Named().Name(), err)
 
 	// First attempt: looking for the blob locally
-	desc, err = repo.app.registry.BlobStatter().Stat(ctx, dgst)
+	desc, err = bs.repo.app.registry.BlobStatter().Stat(ctx, dgst)
 	if err == nil {
 		context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: blob %s exists in the global blob store", dgst.String())
 		// only non-empty layers is wise to check for existence in the image stream.
 		// schema v2 has no empty layers.
 		if !isEmptyDigest(dgst) {
 			// ensure it's referenced inside of corresponding image stream
-			if !imageStreamHasBlob(repo, dgst) {
-				context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: blob %s is neither empty nor referenced in image stream %s", dgst.String(), repo.Named().Name())
+			if !imageStreamHasBlob(bs.repo, dgst) {
+				context.GetLogger(ctx).Debugf("(*blobDescriptorService).Stat: blob %s is neither empty nor referenced in image stream %s", dgst.String(), bs.repo.Named().Name())
 				return distribution.Descriptor{}, distribution.ErrBlobUnknown
 			}
 		}
@@ -97,23 +82,16 @@ func (bs *blobDescriptorService) Stat(ctx context.Context, dgst digest.Digest) (
 
 	if err == distribution.ErrBlobUnknown && remoteBlobAccessCheckEnabledFrom(ctx) {
 		// Second attempt: looking for the blob on a remote server
-		desc, err = repo.remoteBlobGetter.Stat(ctx, dgst)
+		desc, err = bs.repo.remoteBlobGetter.Stat(ctx, dgst)
 	}
 
 	return desc, err
 }
 
 func (bs *blobDescriptorService) Clear(ctx context.Context, dgst digest.Digest) error {
-	repo, found := repositoryFrom(ctx)
-	if !found || repo == nil {
-		err := fmt.Errorf("failed to retrieve repository from context")
-		context.GetLogger(ctx).Error(err)
-		return err
-	}
-
-	repo.cachedLayers.ForgetDigest(dgst, imageapi.DockerImageReference{
-		Namespace: repo.namespace,
-		Name:      repo.name,
+	bs.repo.cachedLayers.ForgetDigest(dgst, imageapi.DockerImageReference{
+		Namespace: bs.repo.namespace,
+		Name:      bs.repo.name,
 	}.Exact())
 	return bs.BlobDescriptorService.Clear(ctx, dgst)
 }
@@ -166,7 +144,7 @@ func imageStreamHasBlob(r *repository, dgst digest.Digest) bool {
 		if _, processed := processedImages[tagEvent.Image]; processed {
 			continue
 		}
-		if imageHasBlob(r, repoCacheName, tagEvent.Image, dgst.String(), !r.config.pullthrough) {
+		if imageHasBlob(r, repoCacheName, tagEvent.Image, dgst.String(), !r.app.extraConfig.Pullthrough.Enabled) {
 			tagName := event2Name[tagEvent]
 			context.GetLogger(r.ctx).Debugf("blob found under istag %s/%s:%s in image %s", r.namespace, r.name, tagName, tagEvent.Image)
 			return logFound(true)