@@ -13,7 +13,7 @@ storage:

 auth:

   openshift:

     realm: openshift

-    

+

     # tokenrealm is a base URL to use for the token-granting registry endpoint.

     # If unspecified, the scheme and host for the token redirect are determined from the incoming request.

     # If specified, a scheme and host must be chosen that all registry clients can resolve and access:

@@ -27,6 +27,7 @@ middleware:

       options:

         acceptschema2: false

         pullthrough: true

+        mirrorpullthrough: true

         enforcequota: false

         projectcachettl: 1m

         blobrepositorycachettl: 10m

@@ -34,7 +34,12 @@ const PruneImagesRecommendedName = "images"

 var (

 	imagesLongDesc = templates.LongDesc(`

-		Prune images no longer needed due to age and/or status

+		Remove image stream tags, images, and image layers by age or usage

+

+		This command removes historical image stream tags, unused images, and unreferenced image

+		layers from the integrated registry. It prefers images that have been directly pushed to

+		the registry, but you may specify --all to include images that were imported (if registry

+		mirroring is enabled).

 		By default, the prune operation performs a dry run making no changes to internal registry. A

 		--confirm flag is needed for changes to be effective.

@@ -51,7 +56,7 @@ var (

 	  %[1]s %[2]s --keep-tag-revisions=3 --keep-younger-than=60m --confirm

 	  # See, what the prune command would delete if we're interested in removing images

-	  # exceeding currently set LimitRanges ('openshift.io/Image')

+	  # exceeding currently set limit ranges ('openshift.io/Image')

 	  %[1]s %[2]s --prune-over-size-limit

 	  # To actually perform the prune operation, the confirm flag must be appended

@@ -70,6 +75,7 @@ type PruneImagesOptions struct {

 	KeepYoungerThan     *time.Duration

 	KeepTagRevisions    *int

 	PruneOverSizeLimit  *bool

+	AllImages           *bool

 	CABundle            string

 	RegistryUrlOverride string

 	Namespace           string

@@ -82,11 +88,13 @@ type PruneImagesOptions struct {

 // NewCmdPruneImages implements the OpenShift cli prune images command.

 func NewCmdPruneImages(f *clientcmd.Factory, parentName, name string, out io.Writer) *cobra.Command {

+	allImages := false

 	opts := &PruneImagesOptions{

 		Confirm:            false,

 		KeepYoungerThan:    &defaultKeepYoungerThan,

 		KeepTagRevisions:   &defaultKeepTagRevisions,

 		PruneOverSizeLimit: &defaultPruneImageOverSizeLimit,

+		AllImages:          &allImages,

 	}

 	cmd := &cobra.Command{

@@ -104,6 +112,7 @@ func NewCmdPruneImages(f *clientcmd.Factory, parentName, name string, out io.Wri

 	}

 	cmd.Flags().BoolVar(&opts.Confirm, "confirm", opts.Confirm, "Specify that image pruning should proceed. Defaults to false, displaying what would be deleted but not actually deleting anything.")

+	cmd.Flags().BoolVar(opts.AllImages, "all", *opts.AllImages, "Include images that were not pushed to the registry but have been mirrored by pullthrough. Requires --registry-url")

 	cmd.Flags().DurationVar(opts.KeepYoungerThan, "keep-younger-than", *opts.KeepYoungerThan, "Specify the minimum age of an image for it to be considered a candidate for pruning.")

 	cmd.Flags().IntVar(opts.KeepTagRevisions, "keep-tag-revisions", *opts.KeepTagRevisions, "Specify the number of image revisions for a tag in an image stream that will be preserved.")

 	cmd.Flags().BoolVar(opts.PruneOverSizeLimit, "prune-over-size-limit", *opts.PruneOverSizeLimit, "Specify if images which are exceeding LimitRanges (see 'openshift.io/Image'), specified in the same namespace, should be considered for pruning. This flag cannot be combined with --keep-younger-than nor --keep-tag-revisions.")

@@ -130,6 +139,11 @@ func (o *PruneImagesOptions) Complete(f *clientcmd.Factory, cmd *cobra.Command,

 			o.KeepTagRevisions = nil

 		}

 	}

+	if *o.AllImages {

+		if len(o.RegistryUrlOverride) == 0 {

+			return kcmdutil.UsageError(cmd, "--registry-url must be specified when --all is true")

+		}

+	}

 	o.Namespace = kapi.NamespaceAll

 	if cmd.Flags().Lookup("namespace").Changed {

 		var err error

@@ -228,6 +242,7 @@ func (o PruneImagesOptions) Run() error {

 		KeepYoungerThan:    o.KeepYoungerThan,

 		KeepTagRevisions:   o.KeepTagRevisions,

 		PruneOverSizeLimit: o.PruneOverSizeLimit,

+		AllImages:          o.AllImages,

 		Images:             allImages,

 		Streams:            allStreams,

 		Pods:               allPods,

@@ -1,8 +1,9 @@

 package server

 import (

+	"io"

 	"net/http"

-	"strconv"

+	"sync"

 	"time"

 	"github.com/docker/distribution"

@@ -23,6 +24,7 @@ type pullthroughBlobStore struct {

 	repo                       *repository

 	digestToStore              map[string]distribution.BlobStore

 	pullFromInsecureRegistries bool

+	mirror                     bool

 }

 var _ distribution.BlobStore = &pullthroughBlobStore{}

@@ -116,30 +118,36 @@ func (r *pullthroughBlobStore) proxyStat(ctx context.Context, retriever importer

 }

 // ServeBlob attempts to serve the requested digest onto w, using a remote proxy store if necessary.

-func (r *pullthroughBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, req *http.Request, dgst digest.Digest) error {

-	store, ok := r.digestToStore[dgst.String()]

+func (pbs *pullthroughBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, req *http.Request, dgst digest.Digest) error {

+	store, ok := pbs.digestToStore[dgst.String()]

 	if !ok {

-		return r.BlobStore.ServeBlob(ctx, w, req, dgst)

+		return pbs.BlobStore.ServeBlob(ctx, w, req, dgst)

 	}

-	desc, err := store.Stat(ctx, dgst)

-	if err != nil {

-		context.GetLogger(ctx).Errorf("failed to stat digest %q: %v", dgst.String(), err)

-		return err

-	}

+	// store the content locally if requested, but ensure only one instance at a time

+	// is storing to avoid excessive local writes

+	if pbs.mirror {

+		mu.Lock()

+		if _, ok = inflight[dgst]; ok {

+			mu.Unlock()

+			context.GetLogger(ctx).Infof("Serving %q while mirroring in background", dgst)

+			_, err := pbs.copyContent(store, ctx, dgst, w, req)

+			return err

+		}

+		inflight[dgst] = struct{}{}

+		mu.Unlock()

-	remoteReader, err := store.Open(ctx, dgst)

-	if err != nil {

-		context.GetLogger(ctx).Errorf("failure to open remote store for digest %q: %v", dgst.String(), err)

-		return err

+		go func(dgst digest.Digest) {

+			context.GetLogger(ctx).Infof("Start background mirroring of %q", dgst)

+			if err := pbs.storeLocal(store, ctx, dgst); err != nil {

+				context.GetLogger(ctx).Errorf("Error committing to storage: %s", err.Error())

+			}

+			context.GetLogger(ctx).Infof("Completed mirroring of %q", dgst)

+		}(dgst)

 	}

-	defer remoteReader.Close()

-	setResponseHeaders(w, desc.Size, desc.MediaType, dgst)

-

-	context.GetLogger(ctx).Infof("serving blob %s of type %s %d bytes long", dgst.String(), desc.MediaType, desc.Size)

-	http.ServeContent(w, req, desc.Digest.String(), time.Time{}, remoteReader)

-	return nil

+	_, err := pbs.copyContent(store, ctx, dgst, w, req)

+	return err

 }

 // Get attempts to fetch the requested blob by digest using a remote proxy store if necessary.

@@ -239,8 +247,72 @@ func identifyCandidateRepositories(is *imageapi.ImageStream, localRegistry strin

 // setResponseHeaders sets the appropriate content serving headers

 func setResponseHeaders(w http.ResponseWriter, length int64, mediaType string, digest digest.Digest) {

-	w.Header().Set("Content-Length", strconv.FormatInt(length, 10))

 	w.Header().Set("Content-Type", mediaType)

 	w.Header().Set("Docker-Content-Digest", digest.String())

 	w.Header().Set("Etag", digest.String())

 }

+

+// inflight tracks currently downloading blobs

+var inflight = make(map[digest.Digest]struct{})

+

+// mu protects inflight

+var mu sync.Mutex

+

+// copyContent attempts to load and serve the provided blob. If req != nil and writer is an instance of http.ResponseWriter,

+// response headers will be set and range requests honored.

+func (pbs *pullthroughBlobStore) copyContent(store distribution.BlobStore, ctx context.Context, dgst digest.Digest, writer io.Writer, req *http.Request) (distribution.Descriptor, error) {

+	desc, err := store.Stat(ctx, dgst)

+	if err != nil {

+		return distribution.Descriptor{}, err

+	}

+

+	remoteReader, err := store.Open(ctx, dgst)

+	if err != nil {

+		return distribution.Descriptor{}, err

+	}

+

+	rw, ok := writer.(http.ResponseWriter)

+	if ok {

+		setResponseHeaders(rw, desc.Size, desc.MediaType, dgst)

+		// serve range requests

+		if req != nil {

+			http.ServeContent(rw, req, desc.Digest.String(), time.Time{}, remoteReader)

+			return desc, nil

+		}

+	}

+

+	if _, err = io.CopyN(writer, remoteReader, desc.Size); err != nil {

+		return distribution.Descriptor{}, err

+	}

+	return desc, nil

+}

+

+// storeLocal retrieves the named blob from the provided store and writes it into the local store.

+func (pbs *pullthroughBlobStore) storeLocal(store distribution.BlobStore, ctx context.Context, dgst digest.Digest) error {

+	defer func() {

+		mu.Lock()

+		delete(inflight, dgst)

+		mu.Unlock()

+	}()

+

+	var desc distribution.Descriptor

+	var err error

+	var bw distribution.BlobWriter

+

+	bw, err = pbs.BlobStore.Create(ctx)

+	if err != nil {

+		return err

+	}

+

+	desc, err = pbs.copyContent(store, ctx, dgst, bw, nil)

+	if err != nil {

+		return err

+	}

+

+	_, err = bw.Commit(ctx, desc)

+	if err != nil {

+		return err

+	}

+

+	return nil

+}

@@ -55,6 +55,12 @@ const (

 	// leaking a blob that is no longer tagged in given repository.

 	BlobRepositoryCacheTTLEnvVar = "REGISTRY_MIDDLEWARE_REPOSITORY_OPENSHIFT_BLOBREPOSITORYCACHETTL"

+	// Pullthrough is a boolean environment variable that controls whether pullthrough is enabled.

+	PullthroughEnvVar = "REGISTRY_MIDDLEWARE_REPOSITORY_OPENSHIFT_PULLTHROUGH"

+

+	// MirrorPullthrough is a boolean environment variable that controls mirroring of blobs on pullthrough.

+	MirrorPullthroughEnvVar = "REGISTRY_MIDDLEWARE_REPOSITORY_OPENSHIFT_MIRRORPULLTHROUGH"

+

 	// Default values

 	defaultDigestToRepositoryCacheSize = 2048

@@ -136,6 +142,8 @@ type repository struct {

 	// if true, the repository will check remote references in the image stream to support pulling "through"

 	// from a remote repository

 	pullthrough bool

+	// mirrorPullthrough will mirror remote blobs into the local repository if set

+	mirrorPullthrough bool

 	// acceptschema2 allows to refuse the manifest schema version 2

 	acceptschema2 bool

 	// blobrepositorycachettl is an eviction timeout for <blob belongs to repository> entries of cachedLayers

@@ -170,7 +178,11 @@ func newRepositoryWithClient(

 	if err != nil {

 		context.GetLogger(ctx).Error(err)

 	}

-	pullthrough, err := getBoolOption("", "pullthrough", false, options)

+	pullthrough, err := getBoolOption(PullthroughEnvVar, "pullthrough", false, options)

+	if err != nil {

+		context.GetLogger(ctx).Error(err)

+	}

+	mirrorPullthrough, err := getBoolOption(MirrorPullthroughEnvVar, "mirrorpullthrough", true, options)

 	if err != nil {

 		context.GetLogger(ctx).Error(err)

 	}

@@ -193,6 +205,7 @@ func newRepositoryWithClient(

 		acceptschema2:          acceptschema2,

 		blobrepositorycachettl: blobrepositorycachettl,

 		pullthrough:            pullthrough,

+		mirrorPullthrough:      mirrorPullthrough,

 		cachedLayers:           cachedLayers,

 	}, nil

 }

@@ -228,6 +241,7 @@ func (r *repository) Blobs(ctx context.Context) distribution.BlobStore {

 			repo:          &repo,

 			digestToStore: make(map[string]distribution.BlobStore),

+			mirror:        r.mirrorPullthrough,

 		}

 	}

@@ -55,6 +55,7 @@ type pruneAlgorithm struct {

 	keepTagRevisions   int

 	pruneOverSizeLimit bool

 	namespace          string

+	allImages          bool

 }

 // ImageDeleter knows how to remove images from OpenShift.

@@ -103,6 +104,9 @@ type PrunerOptions struct {

 	// PruneOverSizeLimit indicates that images exceeding defined limits (openshift.io/Image)

 	// will be considered as candidates for pruning.

 	PruneOverSizeLimit *bool

+	// AllImages considers all images for pruning, not just those pushed directly to the registry.

+	// Requires RegistryURL be set.

+	AllImages *bool

 	// Namespace to be pruned, if specified it should never remove Images.

 	Namespace string

 	// Images is the entire list of images in OpenShift. An image must be in this

@@ -224,9 +228,10 @@ func (*dryRunRegistryPinger) ping(registry string) error {

 // cluster; otherwise, the pruning algorithm might result in incorrect

 // calculations and premature pruning.

 //

-// The ImageDeleter performs the following logic: remove any image containing the

-// annotation openshift.io/image.managed=true that was created at least *n*

-// minutes ago and is *not* currently referenced by:

+// The ImageDeleter performs the following logic:

+//

+// remove any image was created at least *n* minutes ago and is *not* currently

+// referenced by:

 //

 // - any pod created less than *n* minutes ago

 // - any image stream created less than *n* minutes ago

@@ -238,6 +243,9 @@ func (*dryRunRegistryPinger) ping(registry string) error {

 // - any builds

 // - the n most recent tag revisions in an image stream's status.tags

 //

+// including only images with the annotation openshift.io/image.managed=true

+// unless allImages is true.

+//

 // When removing an image, remove all references to the image from all

 // ImageStreams having a reference to the image in `status.tags`.

 //

@@ -252,8 +260,8 @@ func NewPruner(options PrunerOptions) Pruner {

 	if options.PruneOverSizeLimit != nil {

 		pruneOverSizeLimit = fmt.Sprintf("%v", *options.PruneOverSizeLimit)

 	}

-	glog.V(1).Infof("Creating image pruner with keepYoungerThan=%v, keepTagRevisions=%s, pruneOverSizeLimit=%s",

-		options.KeepYoungerThan, keepTagRevisions, pruneOverSizeLimit)

+	glog.V(1).Infof("Creating image pruner with keepYoungerThan=%v, keepTagRevisions=%s, pruneOverSizeLimit=%s allImages=%t",

+		options.KeepYoungerThan, keepTagRevisions, pruneOverSizeLimit, options.AllImages)

 	algorithm := pruneAlgorithm{}

 	if options.KeepYoungerThan != nil {

@@ -265,6 +273,9 @@ func NewPruner(options PrunerOptions) Pruner {

 	if options.PruneOverSizeLimit != nil {

 		algorithm.pruneOverSizeLimit = *options.PruneOverSizeLimit

 	}

+	if options.AllImages != nil {

+		algorithm.allImages = *options.AllImages

+	}

 	algorithm.namespace = options.Namespace

 	g := graph.New()

@@ -302,13 +313,15 @@ func addImagesToGraph(g graph.Graph, images *imageapi.ImageList, algorithm prune

 		glog.V(4).Infof("Examining image %q", image.Name)

-		if image.Annotations == nil {

-			glog.V(4).Infof("Image %q with DockerImageReference %q belongs to an external registry - skipping", image.Name, image.DockerImageReference)

-			continue

-		}

-		if value, ok := image.Annotations[imageapi.ManagedByOpenShiftAnnotation]; !ok || value != "true" {

-			glog.V(4).Infof("Image %q with DockerImageReference %q belongs to an external registry - skipping", image.Name, image.DockerImageReference)

-			continue

+		if !algorithm.allImages {

+			if image.Annotations == nil {

+				glog.V(4).Infof("Image %q with DockerImageReference %q belongs to an external registry - skipping", image.Name, image.DockerImageReference)

+				continue

+			}

+			if value, ok := image.Annotations[imageapi.ManagedByOpenShiftAnnotation]; !ok || value != "true" {

+				glog.V(4).Infof("Image %q with DockerImageReference %q belongs to an external registry - skipping", image.Name, image.DockerImageReference)

+				continue

+			}

 		}

 		age := unversioned.Now().Sub(image.CreationTimestamp.Time)

@@ -105,6 +105,7 @@ os::cmd::try_until_success "curl --max-time 2 --fail --silent 'http://${DOCKER_R

 os::cmd::expect_success "curl -f http://${DOCKER_REGISTRY}/healthz"

 os::cmd::expect_success "dig @${API_HOST} docker-registry.default.local. A"

+registry_pod=$(oc get pod -n default -l deploymentconfig=docker-registry --template='{{(index .items 0).metadata.name}}')

 # Client setup (log in as e2e-user and set 'test' as the default project)

 # This is required to be able to push to the registry!

@@ -147,6 +148,11 @@ os::log::info "Ruby's testing blob digest: $rubyimageblob"

 os::log::info "Docker pullthrough"

 os::cmd::expect_success "oc import-image --confirm --from=mysql:latest mysql:pullthrough"

 os::cmd::expect_success "docker pull ${DOCKER_REGISTRY}/cache/mysql:pullthrough"

+mysqlblob="$(oc get istag -o go-template='{{range .image.dockerImageLayers}}{{if gt .size 1024.}}{{.name}},{{end}}{{end}}' "mysql:pullthrough" | cut -d , -f 1)"

+# directly hit the image to trigger mirroring in case the layer already exists on disk

+os::cmd::expect_success "curl -H 'Authorization: bearer $(oc whoami -t)' 'http://${DOCKER_REGISTRY}/v2/cache/mysql/blobs/${mysqlblob}' 1>/dev/null"

+# verify the blob exists on disk in the registry due to mirroring under .../blobs/sha256/<2 char prefix>/<sha value>

+os::cmd::try_until_success "oc exec --context='${CLUSTER_ADMIN_CONTEXT}' -n default -p ${registry_pod} du /registry | tee '${LOG_DIR}/registry-images.txt' | grep '${mysqlblob:7:100}' | grep blobs"

 os::log::info "Docker registry start with GCS"

 os::cmd::expect_failure_and_text "docker run -e REGISTRY_STORAGE=\"gcs: {}\" openshift/origin-docker-registry:${TAG}" "No bucket parameter provided"

@@ -519,23 +525,34 @@ os::cmd::expect_success "docker tag gcr.io/google_containers/pause ${DOCKER_REGI

 os::cmd::expect_success "docker push ${DOCKER_REGISTRY}/cache/prune"

 # record the storage before pruning

-registry_pod=$(oc get pod -l deploymentconfig=docker-registry --template='{{(index .items 0).metadata.name}}')

+registry_pod=$(oc get pod -n default -l deploymentconfig=docker-registry --template='{{(index .items 0).metadata.name}}')

 os::cmd::expect_success "oc exec -p ${registry_pod} du /registry > '${LOG_DIR}/prune-images.before.txt'"

 # set up pruner user

-os::cmd::expect_success 'oadm policy add-cluster-role-to-user system:image-pruner e2e-pruner'

-os::cmd::try_until_text 'oadm policy who-can list images' 'e2e-pruner'

-os::cmd::expect_success 'oc login -u e2e-pruner -p pass'

+os::cmd::expect_success 'oadm policy add-cluster-role-to-user system:image-pruner system:serviceaccount:cache:builder'

+os::cmd::try_until_text 'oadm policy who-can list images' 'system:serviceaccount:cache:builder'

 # run image pruning

-os::cmd::expect_success_and_not_text "oadm prune images --keep-younger-than=0 --keep-tag-revisions=1 --confirm" 'error'

+os::cmd::expect_success_and_not_text "oadm prune images --token=$(oc sa get-token builder -n cache) --keep-younger-than=0 --keep-tag-revisions=1 --confirm" 'error'

-os::cmd::expect_success "oc project ${CLUSTER_ADMIN_CONTEXT}"

 # record the storage after pruning

 os::cmd::expect_success "oc exec -p ${registry_pod} du /registry > '${LOG_DIR}/prune-images.after.txt'"

 # make sure there were changes to the registry's storage

 os::cmd::expect_code "diff ${LOG_DIR}/prune-images.before.txt ${LOG_DIR}/prune-images.after.txt" 1

+

+# prune a mirror, external image that is no longer referenced

+os::cmd::expect_success "oc import-image nginx --confirm -n cache"

+nginxblob="$(oc get istag -o go-template='{{range .image.dockerImageLayers}}{{if gt .size 1024.}}{{.name}},{{end}}{{end}}' "nginx:latest" -n cache | cut -d , -f 1)"

+# directly hit the image to trigger mirroring in case the layer already exists on disk

+os::cmd::expect_success "curl -H 'Authorization: bearer $(oc sa get-token builder -n cache)' 'http://${DOCKER_REGISTRY}/v2/cache/nginx/blobs/${nginxblob}' 1>/dev/null"

+# verify the blob exists on disk in the registry due to mirroring under .../blobs/sha256/<2 char prefix>/<sha value>

+os::cmd::try_until_success "oc exec --context='${CLUSTER_ADMIN_CONTEXT}' -n default -p ${registry_pod} du /registry | tee '${LOG_DIR}/registry-images.txt' | grep '${nginxblob:7:100}' | grep blobs"

+os::cmd::expect_success "oc delete is nginx -n cache"

+os::cmd::expect_success "oc exec -p ${registry_pod} du /registry > '${LOG_DIR}/prune-images.before.txt'"

+os::cmd::expect_success_and_not_text "oadm prune images --token=$(oc sa get-token builder -n cache) --keep-younger-than=0 --confirm --all --registry-url=${DOCKER_REGISTRY}" 'error'

+os::cmd::expect_success "oc exec -p ${registry_pod} du /registry > '${LOG_DIR}/prune-images.after.txt'"

+os::cmd::expect_code "diff ${LOG_DIR}/prune-images.before.txt ${LOG_DIR}/prune-images.after.txt" 1

 os::log::info "Validated image pruning"

 # with registry's re-deployment we loose all the blobs stored in its storage until now