1. docker
  2. Commit 81f7b1f1e50252e9fdd52111f0899d59f19ecb87
Browse code

Different number of retries for layers of different sizes

Classify blobs into three categories based on size.
Use a very limited number of mount attempts and no existence check for
small blobs. Use more attempts for bigger blobs.

Also remember blob associations during layer existence check.

Blob digests are now checked in the target repository from newest to
latest. If the blob exists and the metadata entry does not, it will be
created. If the blob is not found, the metadata entry will be removed.

Signed-off-by: Michal Minář <miminar@redhat.com>

Michal Minář authored on 2016/09/16 20:58:36
Showing 2 changed files
distribution/push_v2.go
History View file @ 81f7b1f
... ... 
@@ -29,7 +29,10 @@ import (
29 29 
 	"github.com/docker/docker/registry"
30 30 
 )
31 31
32 
-const maxRepositoryMountAttempts = 4
32 
+const (
33 
+	smallLayerMaximumSize  = 100 * (1 << 10) // 100KB
34 
+	middleLayerMaximumSize = 10 * (1 << 20)  // 10MB
35 
+)
33 36
34 37 
 // PushResult contains the tag, manifest digest, and manifest size from the
35 38 
 // push. It's used to signal this information to the trust code in the client
... ... 
@@ -158,6 +161,7 @@ func (p *v2Pusher) pushV2Tag(ctx context.Context, ref reference.NamedTagged, ima
158 158 
 	for i := 0; i < len(img.RootFS.DiffIDs); i++ {
159 159 
 		descriptor := descriptorTemplate
160 160 
 		descriptor.layer = l
161 
+		descriptor.checkedDigests = make(map[digest.Digest]struct{})
161 162 
 		descriptors = append(descriptors, &descriptor)
162 163
163 164 
 		l = l.Parent()
... ... 
@@ -250,6 +254,8 @@ type v2PushDescriptor struct {
250 250 
 	repo              distribution.Repository
251 251 
 	pushState         *pushState
252 252 
 	remoteDescriptor  distribution.Descriptor
253 
+	// a set of digests whose presence has been checked in a target repository
254 
+	checkedDigests map[digest.Digest]struct{}
253 255 
 }
254 256
255 257 
 func (pd *v2PushDescriptor) Key() string {
... ... 
@@ -284,25 +290,18 @@ func (pd *v2PushDescriptor) Upload(ctx context.Context, progressOutput progress.
284 284 
 	}
285 285 
 	pd.pushState.Unlock()
286 286
287 
+	maxMountAttempts, maxExistenceChecks, checkOtherRepositories := getMaxMountAndExistenceCheckAttempts(pd.layer)
288 
+
287 289 
 	// Do we have any metadata associated with this layer's DiffID?
288 290 
 	v2Metadata, err := pd.v2MetadataService.GetMetadata(diffID)
289 291 
 	if err == nil {
290 
-		descriptor, exists, err := layerAlreadyExists(ctx, v2Metadata, pd.repoInfo, pd.repo, pd.pushState)
291 
-		if err != nil {
292 
-			progress.Update(progressOutput, pd.ID(), "Image push failed")
293 
-			return distribution.Descriptor{}, retryOnError(err)
294 
-		}
295 
-		if exists {
296 
-			progress.Update(progressOutput, pd.ID(), "Layer already exists")
297 
-			pd.pushState.Lock()
298 
-			pd.pushState.remoteLayers[diffID] = descriptor
299 
-			pd.pushState.Unlock()
300 
-			return descriptor, nil
292 
+		// check for blob existence in the target repository if we have a mapping with it
293 
+		descriptor, exists, err := pd.layerAlreadyExists(ctx, progressOutput, diffID, false, 1, v2Metadata)
294 
+		if exists || err != nil {
295 
+			return descriptor, err
301 296 
 		}
302 297 
 	}
303 298
304 
-	logrus.Debugf("Pushing layer: %s", diffID)
305 
-
306 299 
 	// if digest was empty or not saved, or if blob does not exist on the remote repository,
307 300 
 	// then push the blob.
308 301 
 	bs := pd.repo.Blobs(ctx)
... ... 
@@ -310,7 +309,7 @@ func (pd *v2PushDescriptor) Upload(ctx context.Context, progressOutput progress.
310 310 
 	var layerUpload distribution.BlobWriter
311 311
312 312 
 	// Attempt to find another repository in the same registry to mount the layer from to avoid an unnecessary upload
313 
-	candidates := getRepositoryMountCandidates(pd.repoInfo, pd.hmacKey, maxRepositoryMountAttempts, v2Metadata)
313 
+	candidates := getRepositoryMountCandidates(pd.repoInfo, pd.hmacKey, maxMountAttempts, v2Metadata)
314 314 
 	for _, mountCandidate := range candidates {
315 315 
 		logrus.Debugf("attempting to mount layer %s (%s) from %s", diffID, mountCandidate.Digest, mountCandidate.SourceRepository)
316 316 
 		createOpts := []distribution.BlobCreateOption{}
... ... 
@@ -386,6 +385,15 @@ func (pd *v2PushDescriptor) Upload(ctx context.Context, progressOutput progress.
386 386 
 		}
387 387 
 	}
388 388
389 
+	if maxExistenceChecks-len(pd.checkedDigests) > 0 {
390 
+		// do additional layer existence checks with other known digests if any
391 
+		descriptor, exists, err := pd.layerAlreadyExists(ctx, progressOutput, diffID, checkOtherRepositories, maxExistenceChecks-len(pd.checkedDigests), v2Metadata)
392 
+		if exists || err != nil {
393 
+			return descriptor, err
394 
+		}
395 
+	}
396 
+
397 
+	logrus.Debugf("Pushing layer: %s", diffID)
389 398 
 	if layerUpload == nil {
390 399 
 		layerUpload, err = bs.Create(ctx)
391 400 
 		if err != nil {
... ... 
@@ -400,12 +408,6 @@ func (pd *v2PushDescriptor) Upload(ctx context.Context, progressOutput progress.
400 400 
 		return desc, err
401 401 
 	}
402 402
403 
-	pd.pushState.Lock()
404 
-	// If Commit succeeded, that's an indication that the remote registry speaks the v2 protocol.
405 
-	pd.pushState.confirmedV2 = true
406 
-	pd.pushState.remoteLayers[diffID] = desc
407 
-	pd.pushState.Unlock()
408 
-
409 403 
 	return desc, nil
410 404 
 }
411 405
... ... 
@@ -463,34 +465,130 @@ func (pd *v2PushDescriptor) uploadUsingSession(
463 463 
 		return distribution.Descriptor{}, xfer.DoNotRetry{Err: err}
464 464 
 	}
465 465
466 
-	return distribution.Descriptor{
466 
+	desc := distribution.Descriptor{
467 467 
 		Digest:    pushDigest,
468 468 
 		MediaType: schema2.MediaTypeLayer,
469 469 
 		Size:      nn,
470 
-	}, nil
470 
+	}
471 
+
472 
+	pd.pushState.Lock()
473 
+	// If Commit succeeded, that's an indication that the remote registry speaks the v2 protocol.
474 
+	pd.pushState.confirmedV2 = true
475 
+	pd.pushState.remoteLayers[diffID] = desc
476 
+	pd.pushState.Unlock()
477 
+
478 
+	return desc, nil
471 479 
 }
472 480
473 
-// layerAlreadyExists checks if the registry already know about any of the
474 
-// metadata passed in the "metadata" slice. If it finds one that the registry
475 
-// knows about, it returns the known digest and "true".
476 
-func layerAlreadyExists(ctx context.Context, metadata []metadata.V2Metadata, repoInfo reference.Named, repo distribution.Repository, pushState *pushState) (distribution.Descriptor, bool, error) {
477 
-	for _, meta := range metadata {
478 
-		// Only check blobsums that are known to this repository or have an unknown source
479 
-		if meta.SourceRepository != "" && meta.SourceRepository != repoInfo.FullName() {
481 
+// layerAlreadyExists checks if the registry already knows about any of the metadata passed in the "metadata"
482 
+// slice. If it finds one that the registry knows about, it returns the known digest and "true". If
483 
+// "checkOtherRepositories" is true, stat will be performed also with digests mapped to any other repository
484 
+// (not just the target one).
485 
+func (pd *v2PushDescriptor) layerAlreadyExists(
486 
+	ctx context.Context,
487 
+	progressOutput progress.Output,
488 
+	diffID layer.DiffID,
489 
+	checkOtherRepositories bool,
490 
+	maxExistenceCheckAttempts int,
491 
+	v2Metadata []metadata.V2Metadata,
492 
+) (desc distribution.Descriptor, exists bool, err error) {
493 
+	// filter the metadata
494 
+	candidates := []metadata.V2Metadata{}
495 
+	for _, meta := range v2Metadata {
496 
+		if len(meta.SourceRepository) > 0 && !checkOtherRepositories && meta.SourceRepository != pd.repoInfo.FullName() {
497 
+			continue
498 
+		}
499 
+		candidates = append(candidates, meta)
500 
+	}
501 
+	// sort the candidates by similarity
502 
+	sortV2MetadataByLikenessAndAge(pd.repoInfo, pd.hmacKey, candidates)
503 
+
504 
+	digestToMetadata := make(map[digest.Digest]*metadata.V2Metadata)
505 
+	// an array of unique blob digests ordered from the best mount candidates to worst
506 
+	layerDigests := []digest.Digest{}
507 
+	for i := 0; i < len(candidates); i++ {
508 
+		if len(layerDigests) >= maxExistenceCheckAttempts {
509 
+			break
510 
+		}
511 
+		meta := &candidates[i]
512 
+		if _, exists := digestToMetadata[meta.Digest]; exists {
513 
+			// keep reference just to the first mapping (the best mount candidate)
514 
+			continue
515 
+		}
516 
+		if _, exists := pd.checkedDigests[meta.Digest]; exists {
517 
+			// existence of this digest has already been tested
480 518 
 			continue
481 519 
 		}
482 
-		descriptor, err := repo.Blobs(ctx).Stat(ctx, meta.Digest)
520 
+		digestToMetadata[meta.Digest] = meta
521 
+		layerDigests = append(layerDigests, meta.Digest)
522 
+	}
523 
+
524 
+	for _, dgst := range layerDigests {
525 
+		meta := digestToMetadata[dgst]
526 
+		logrus.Debugf("Checking for presence of layer %s (%s) in %s", diffID, dgst, pd.repoInfo.FullName())
527 
+		desc, err = pd.repo.Blobs(ctx).Stat(ctx, dgst)
528 
+		pd.checkedDigests[meta.Digest] = struct{}{}
483 529 
 		switch err {
484 530 
 		case nil:
485 
-			descriptor.MediaType = schema2.MediaTypeLayer
486 
-			return descriptor, true, nil
531 
+			if m, ok := digestToMetadata[desc.Digest]; !ok || m.SourceRepository != pd.repoInfo.FullName() || !metadata.CheckV2MetadataHMAC(m, pd.hmacKey) {
532 
+				// cache mapping from this layer's DiffID to the blobsum
533 
+				if err := pd.v2MetadataService.TagAndAdd(diffID, pd.hmacKey, metadata.V2Metadata{
534 
+					Digest:           desc.Digest,
535 
+					SourceRepository: pd.repoInfo.FullName(),
536 
+				}); err != nil {
537 
+					return distribution.Descriptor{}, false, xfer.DoNotRetry{Err: err}
538 
+				}
539 
+			}
540 
+			desc.MediaType = schema2.MediaTypeLayer
541 
+			exists = true
542 
+			break
487 543 
 		case distribution.ErrBlobUnknown:
488 
-			// nop
544 
+			if meta.SourceRepository == pd.repoInfo.FullName() {
545 
+				// remove the mapping to the target repository
546 
+				pd.v2MetadataService.Remove(*meta)
547 
+			}
489 548 
 		default:
490 
-			return distribution.Descriptor{}, false, err
549 
+			progress.Update(progressOutput, pd.ID(), "Image push failed")
550 
+			return desc, false, retryOnError(err)
491 551 
 		}
492 552 
 	}
493 
-	return distribution.Descriptor{}, false, nil
553 
+
554 
+	if exists {
555 
+		progress.Update(progressOutput, pd.ID(), "Layer already exists")
556 
+		pd.pushState.Lock()
557 
+		pd.pushState.remoteLayers[diffID] = desc
558 
+		pd.pushState.Unlock()
559 
+	}
560 
+
561 
+	return desc, exists, nil
562 
+}
563 
+
564 
+// getMaxMountAndExistenceCheckAttempts returns a maximum number of cross repository mount attempts from
565 
+// source repositories of target registry, maximum number of layer existence checks performed on the target
566 
+// repository and whether the check shall be done also with digests mapped to different repositories. The
567 
+// decision is based on layer size. The smaller the layer, the fewer attempts shall be made because the cost
568 
+// of upload does not outweigh a latency.
569 
+func getMaxMountAndExistenceCheckAttempts(layer layer.Layer) (maxMountAttempts, maxExistenceCheckAttempts int, checkOtherRepositories bool) {
570 
+	size, err := layer.DiffSize()
571 
+	switch {
572 
+	// big blob
573 
+	case size > middleLayerMaximumSize:
574 
+		// 1st attempt to mount the blob few times
575 
+		// 2nd few existence checks with digests associated to any repository
576 
+		// then fallback to upload
577 
+		return 4, 3, true
578 
+
579 
+	// middle sized blobs; if we could not get the size, assume we deal with middle sized blob
580 
+	case size > smallLayerMaximumSize, err != nil:
581 
+		// 1st attempt to mount blobs of average size few times
582 
+		// 2nd try at most 1 existence check if there's an existing mapping to the target repository
583 
+		// then fallback to upload
584 
+		return 3, 1, false
585 
+
586 
+	// small blobs, do a minimum number of checks
587 
+	default:
588 
+		return 1, 1, false
589 
+	}
494 590 
 }
495 591
496 592 
 // getRepositoryMountCandidates returns an array of v2 metadata items belonging to the given registry. The
distribution/push_v2_test.go
History View file @ 81f7b1f
... ... 
@@ -1,11 +1,18 @@
1 1 
 package distribution
2 2
3 3 
 import (
4 
+	"net/http"
4 5 
 	"reflect"
5 6 
 	"testing"
6 7
8 
+	"github.com/docker/distribution"
9 
+	"github.com/docker/distribution/context"
7 10 
 	"github.com/docker/distribution/digest"
11 
+	"github.com/docker/distribution/manifest/schema2"
12 
+	distreference "github.com/docker/distribution/reference"
8 13 
 	"github.com/docker/docker/distribution/metadata"
14 
+	"github.com/docker/docker/layer"
15 
+	"github.com/docker/docker/pkg/progress"
9 16 
 	"github.com/docker/docker/reference"
10 17 
 )
11 18
... ... 
@@ -136,6 +143,315 @@ func TestGetRepositoryMountCandidates(t *testing.T) {
136 136 
 	}
137 137 
 }
138 138
139 
+func TestLayerAlreadyExists(t *testing.T) {
140 
+	for _, tc := range []struct {
141 
+		name                   string
142 
+		metadata               []metadata.V2Metadata
143 
+		targetRepo             string
144 
+		hmacKey                string
145 
+		maxExistenceChecks     int
146 
+		checkOtherRepositories bool
147 
+		remoteBlobs            map[digest.Digest]distribution.Descriptor
148 
+		remoteErrors           map[digest.Digest]error
149 
+		expectedDescriptor     distribution.Descriptor
150 
+		expectedExists         bool
151 
+		expectedError          error
152 
+		expectedRequests       []string
153 
+		expectedAdditions      []metadata.V2Metadata
154 
+		expectedRemovals       []metadata.V2Metadata
155 
+	}{
156 
+		{
157 
+			name:                   "empty metadata",
158 
+			targetRepo:             "busybox",
159 
+			maxExistenceChecks:     3,
160 
+			checkOtherRepositories: true,
161 
+		},
162 
+		{
163 
+			name:               "single not existent metadata",
164 
+			targetRepo:         "busybox",
165 
+			metadata:           []metadata.V2Metadata{{Digest: digest.Digest("pear"), SourceRepository: "docker.io/library/busybox"}},
166 
+			maxExistenceChecks: 3,
167 
+			expectedRequests:   []string{"pear"},
168 
+			expectedRemovals:   []metadata.V2Metadata{{Digest: digest.Digest("pear"), SourceRepository: "docker.io/library/busybox"}},
169 
+		},
170 
+		{
171 
+			name:               "access denied",
172 
+			targetRepo:         "busybox",
173 
+			maxExistenceChecks: 1,
174 
+			metadata:           []metadata.V2Metadata{{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/busybox"}},
175 
+			remoteErrors:       map[digest.Digest]error{digest.Digest("apple"): distribution.ErrAccessDenied},
176 
+			expectedError:      distribution.ErrAccessDenied,
177 
+			expectedRequests:   []string{"apple"},
178 
+		},
179 
+		{
180 
+			name:               "not matching reposies",
181 
+			targetRepo:         "busybox",
182 
+			maxExistenceChecks: 3,
183 
+			metadata: []metadata.V2Metadata{
184 
+				{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/hello-world"},
185 
+				{Digest: digest.Digest("orange"), SourceRepository: "docker.io/library/busybox/subapp"},
186 
+				{Digest: digest.Digest("pear"), SourceRepository: "docker.io/busybox"},
187 
+				{Digest: digest.Digest("plum"), SourceRepository: "busybox"},
188 
+				{Digest: digest.Digest("banana"), SourceRepository: "127.0.0.1/busybox"},
189 
+			},
190 
+		},
191 
+		{
192 
+			name:                   "check other repositories",
193 
+			targetRepo:             "busybox",
194 
+			maxExistenceChecks:     10,
195 
+			checkOtherRepositories: true,
196 
+			metadata: []metadata.V2Metadata{
197 
+				{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/hello-world"},
198 
+				{Digest: digest.Digest("orange"), SourceRepository: "docker.io/library/busybox/subapp"},
199 
+				{Digest: digest.Digest("pear"), SourceRepository: "docker.io/busybox"},
200 
+				{Digest: digest.Digest("plum"), SourceRepository: "busybox"},
201 
+				{Digest: digest.Digest("banana"), SourceRepository: "127.0.0.1/busybox"},
202 
+			},
203 
+			expectedRequests: []string{"plum", "pear", "apple", "orange", "banana"},
204 
+		},
205 
+		{
206 
+			name:               "find existing blob",
207 
+			targetRepo:         "busybox",
208 
+			metadata:           []metadata.V2Metadata{{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/busybox"}},
209 
+			maxExistenceChecks: 3,
210 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("apple"): {Digest: digest.Digest("apple")}},
211 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("apple"), MediaType: schema2.MediaTypeLayer},
212 
+			expectedExists:     true,
213 
+			expectedRequests:   []string{"apple"},
214 
+		},
215 
+		{
216 
+			name:               "find existing blob with different hmac",
217 
+			targetRepo:         "busybox",
218 
+			metadata:           []metadata.V2Metadata{{SourceRepository: "docker.io/library/busybox", Digest: digest.Digest("apple"), HMAC: "dummyhmac"}},
219 
+			maxExistenceChecks: 3,
220 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("apple"): {Digest: digest.Digest("apple")}},
221 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("apple"), MediaType: schema2.MediaTypeLayer},
222 
+			expectedExists:     true,
223 
+			expectedRequests:   []string{"apple"},
224 
+			expectedAdditions:  []metadata.V2Metadata{{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/busybox"}},
225 
+		},
226 
+		{
227 
+			name:               "overwrite media types",
228 
+			targetRepo:         "busybox",
229 
+			metadata:           []metadata.V2Metadata{{Digest: digest.Digest("apple"), SourceRepository: "docker.io/library/busybox"}},
230 
+			hmacKey:            "key",
231 
+			maxExistenceChecks: 3,
232 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("apple"): {Digest: digest.Digest("apple"), MediaType: "custom-media-type"}},
233 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("apple"), MediaType: schema2.MediaTypeLayer},
234 
+			expectedExists:     true,
235 
+			expectedRequests:   []string{"apple"},
236 
+			expectedAdditions:  []metadata.V2Metadata{taggedMetadata("key", "apple", "docker.io/library/busybox")},
237 
+		},
238 
+		{
239 
+			name:       "find existing blob among many",
240 
+			targetRepo: "127.0.0.1/myapp",
241 
+			hmacKey:    "key",
242 
+			metadata: []metadata.V2Metadata{
243 
+				taggedMetadata("someotherkey", "pear", "127.0.0.1/myapp"),
244 
+				taggedMetadata("key", "apple", "127.0.0.1/myapp"),
245 
+				taggedMetadata("", "plum", "127.0.0.1/myapp"),
246 
+			},
247 
+			maxExistenceChecks: 3,
248 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("pear"): {Digest: digest.Digest("pear")}},
249 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("pear"), MediaType: schema2.MediaTypeLayer},
250 
+			expectedExists:     true,
251 
+			expectedRequests:   []string{"apple", "plum", "pear"},
252 
+			expectedAdditions:  []metadata.V2Metadata{taggedMetadata("key", "pear", "127.0.0.1/myapp")},
253 
+			expectedRemovals: []metadata.V2Metadata{
254 
+				taggedMetadata("key", "apple", "127.0.0.1/myapp"),
255 
+				{Digest: digest.Digest("plum"), SourceRepository: "127.0.0.1/myapp"},
256 
+			},
257 
+		},
258 
+		{
259 
+			name:       "reach maximum existence checks",
260 
+			targetRepo: "user/app",
261 
+			metadata: []metadata.V2Metadata{
262 
+				{Digest: digest.Digest("pear"), SourceRepository: "docker.io/user/app"},
263 
+				{Digest: digest.Digest("apple"), SourceRepository: "docker.io/user/app"},
264 
+				{Digest: digest.Digest("plum"), SourceRepository: "docker.io/user/app"},
265 
+				{Digest: digest.Digest("banana"), SourceRepository: "docker.io/user/app"},
266 
+			},
267 
+			maxExistenceChecks: 3,
268 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("pear"): {Digest: digest.Digest("pear")}},
269 
+			expectedExists:     false,
270 
+			expectedRequests:   []string{"banana", "plum", "apple"},
271 
+			expectedRemovals: []metadata.V2Metadata{
272 
+				{Digest: digest.Digest("banana"), SourceRepository: "docker.io/user/app"},
273 
+				{Digest: digest.Digest("plum"), SourceRepository: "docker.io/user/app"},
274 
+				{Digest: digest.Digest("apple"), SourceRepository: "docker.io/user/app"},
275 
+			},
276 
+		},
277 
+		{
278 
+			name:       "zero allowed existence checks",
279 
+			targetRepo: "user/app",
280 
+			metadata: []metadata.V2Metadata{
281 
+				{Digest: digest.Digest("pear"), SourceRepository: "docker.io/user/app"},
282 
+				{Digest: digest.Digest("apple"), SourceRepository: "docker.io/user/app"},
283 
+				{Digest: digest.Digest("plum"), SourceRepository: "docker.io/user/app"},
284 
+				{Digest: digest.Digest("banana"), SourceRepository: "docker.io/user/app"},
285 
+			},
286 
+			maxExistenceChecks: 0,
287 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("pear"): {Digest: digest.Digest("pear")}},
288 
+		},
289 
+		{
290 
+			name:       "stat single digest just once",
291 
+			targetRepo: "busybox",
292 
+			metadata: []metadata.V2Metadata{
293 
+				taggedMetadata("key1", "pear", "docker.io/library/busybox"),
294 
+				taggedMetadata("key2", "apple", "docker.io/library/busybox"),
295 
+				taggedMetadata("key3", "apple", "docker.io/library/busybox"),
296 
+			},
297 
+			maxExistenceChecks: 3,
298 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("pear"): {Digest: digest.Digest("pear")}},
299 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("pear"), MediaType: schema2.MediaTypeLayer},
300 
+			expectedExists:     true,
301 
+			expectedRequests:   []string{"apple", "pear"},
302 
+			expectedAdditions:  []metadata.V2Metadata{{Digest: digest.Digest("pear"), SourceRepository: "docker.io/library/busybox"}},
303 
+			expectedRemovals:   []metadata.V2Metadata{taggedMetadata("key3", "apple", "docker.io/library/busybox")},
304 
+		},
305 
+		{
306 
+			name:       "stop on first error",
307 
+			targetRepo: "user/app",
308 
+			hmacKey:    "key",
309 
+			metadata: []metadata.V2Metadata{
310 
+				taggedMetadata("key", "banana", "docker.io/user/app"),
311 
+				taggedMetadata("key", "orange", "docker.io/user/app"),
312 
+				taggedMetadata("key", "plum", "docker.io/user/app"),
313 
+			},
314 
+			maxExistenceChecks: 3,
315 
+			remoteErrors:       map[digest.Digest]error{"orange": distribution.ErrAccessDenied},
316 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("apple"): {}},
317 
+			expectedError:      distribution.ErrAccessDenied,
318 
+			expectedRequests:   []string{"plum", "orange"},
319 
+			expectedRemovals:   []metadata.V2Metadata{taggedMetadata("key", "plum", "docker.io/user/app")},
320 
+		},
321 
+		{
322 
+			name:       "remove outdated metadata",
323 
+			targetRepo: "docker.io/user/app",
324 
+			metadata: []metadata.V2Metadata{
325 
+				{Digest: digest.Digest("plum"), SourceRepository: "docker.io/library/busybox"},
326 
+				{Digest: digest.Digest("orange"), SourceRepository: "docker.io/user/app"},
327 
+			},
328 
+			maxExistenceChecks: 3,
329 
+			remoteErrors:       map[digest.Digest]error{"orange": distribution.ErrBlobUnknown},
330 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("plum"): {}},
331 
+			expectedExists:     false,
332 
+			expectedRequests:   []string{"orange"},
333 
+			expectedRemovals:   []metadata.V2Metadata{{Digest: digest.Digest("orange"), SourceRepository: "docker.io/user/app"}},
334 
+		},
335 
+		{
336 
+			name:       "missing SourceRepository",
337 
+			targetRepo: "busybox",
338 
+			metadata: []metadata.V2Metadata{
339 
+				{Digest: digest.Digest("1")},
340 
+				{Digest: digest.Digest("3")},
341 
+				{Digest: digest.Digest("2")},
342 
+			},
343 
+			maxExistenceChecks: 3,
344 
+			expectedExists:     false,
345 
+			expectedRequests:   []string{"2", "3", "1"},
346 
+		},
347 
+
348 
+		{
349 
+			name:       "with and without SourceRepository",
350 
+			targetRepo: "busybox",
351 
+			metadata: []metadata.V2Metadata{
352 
+				{Digest: digest.Digest("1")},
353 
+				{Digest: digest.Digest("2"), SourceRepository: "docker.io/library/busybox"},
354 
+				{Digest: digest.Digest("3")},
355 
+			},
356 
+			remoteBlobs:        map[digest.Digest]distribution.Descriptor{digest.Digest("1"): {Digest: digest.Digest("1")}},
357 
+			maxExistenceChecks: 3,
358 
+			expectedDescriptor: distribution.Descriptor{Digest: digest.Digest("1"), MediaType: schema2.MediaTypeLayer},
359 
+			expectedExists:     true,
360 
+			expectedRequests:   []string{"2", "3", "1"},
361 
+			expectedAdditions:  []metadata.V2Metadata{{Digest: digest.Digest("1"), SourceRepository: "docker.io/library/busybox"}},
362 
+			expectedRemovals: []metadata.V2Metadata{
363 
+				{Digest: digest.Digest("2"), SourceRepository: "docker.io/library/busybox"},
364 
+			},
365 
+		},
366 
+	} {
367 
+		repoInfo, err := reference.ParseNamed(tc.targetRepo)
368 
+		if err != nil {
369 
+			t.Fatalf("[%s] failed to parse reference name: %v", tc.name, err)
370 
+		}
371 
+		repo := &mockRepo{
372 
+			t:        t,
373 
+			errors:   tc.remoteErrors,
374 
+			blobs:    tc.remoteBlobs,
375 
+			requests: []string{},
376 
+		}
377 
+		ctx := context.Background()
378 
+		ms := &mockV2MetadataService{}
379 
+		pd := &v2PushDescriptor{
380 
+			hmacKey:           []byte(tc.hmacKey),
381 
+			repoInfo:          repoInfo,
382 
+			layer:             layer.EmptyLayer,
383 
+			repo:              repo,
384 
+			v2MetadataService: ms,
385 
+			pushState:         &pushState{remoteLayers: make(map[layer.DiffID]distribution.Descriptor)},
386 
+			checkedDigests:    make(map[digest.Digest]struct{}),
387 
+		}
388 
+
389 
+		desc, exists, err := pd.layerAlreadyExists(ctx, &progressSink{t}, layer.EmptyLayer.DiffID(), tc.checkOtherRepositories, tc.maxExistenceChecks, tc.metadata)
390 
+
391 
+		if !reflect.DeepEqual(desc, tc.expectedDescriptor) {
392 
+			t.Errorf("[%s] got unexpected descriptor: %#+v != %#+v", tc.name, desc, tc.expectedDescriptor)
393 
+		}
394 
+		if exists != tc.expectedExists {
395 
+			t.Errorf("[%s] got unexpected exists: %t != %t", tc.name, exists, tc.expectedExists)
396 
+		}
397 
+		if !reflect.DeepEqual(err, tc.expectedError) {
398 
+			t.Errorf("[%s] got unexpected error: %#+v != %#+v", tc.name, err, tc.expectedError)
399 
+		}
400 
+
401 
+		if len(repo.requests) != len(tc.expectedRequests) {
402 
+			t.Errorf("[%s] got unexpected number of requests: %d != %d", tc.name, len(repo.requests), len(tc.expectedRequests))
403 
+		}
404 
+		for i := 0; i < len(repo.requests) && i < len(tc.expectedRequests); i++ {
405 
+			if repo.requests[i] != tc.expectedRequests[i] {
406 
+				t.Errorf("[%s] request %d does not match expected: %q != %q", tc.name, i, repo.requests[i], tc.expectedRequests[i])
407 
+			}
408 
+		}
409 
+		for i := len(repo.requests); i < len(tc.expectedRequests); i++ {
410 
+			t.Errorf("[%s] missing expected request at position %d (%q)", tc.name, i, tc.expectedRequests[i])
411 
+		}
412 
+		for i := len(tc.expectedRequests); i < len(repo.requests); i++ {
413 
+			t.Errorf("[%s] got unexpected request at position %d (%q)", tc.name, i, repo.requests[i])
414 
+		}
415 
+
416 
+		if len(ms.added) != len(tc.expectedAdditions) {
417 
+			t.Errorf("[%s] got unexpected number of additions: %d != %d", tc.name, len(ms.added), len(tc.expectedAdditions))
418 
+		}
419 
+		for i := 0; i < len(ms.added) && i < len(tc.expectedAdditions); i++ {
420 
+			if ms.added[i] != tc.expectedAdditions[i] {
421 
+				t.Errorf("[%s] added metadata at %d does not match expected: %q != %q", tc.name, i, ms.added[i], tc.expectedAdditions[i])
422 
+			}
423 
+		}
424 
+		for i := len(ms.added); i < len(tc.expectedAdditions); i++ {
425 
+			t.Errorf("[%s] missing expected addition at position %d (%q)", tc.name, i, tc.expectedAdditions[i])
426 
+		}
427 
+		for i := len(tc.expectedAdditions); i < len(ms.added); i++ {
428 
+			t.Errorf("[%s] unexpected metadata addition at position %d (%q)", tc.name, i, ms.added[i])
429 
+		}
430 
+
431 
+		if len(ms.removed) != len(tc.expectedRemovals) {
432 
+			t.Errorf("[%s] got unexpected number of removals: %d != %d", tc.name, len(ms.removed), len(tc.expectedRemovals))
433 
+		}
434 
+		for i := 0; i < len(ms.removed) && i < len(tc.expectedRemovals); i++ {
435 
+			if ms.removed[i] != tc.expectedRemovals[i] {
436 
+				t.Errorf("[%s] removed metadata at %d does not match expected: %q != %q", tc.name, i, ms.removed[i], tc.expectedRemovals[i])
437 
+			}
438 
+		}
439 
+		for i := len(ms.removed); i < len(tc.expectedRemovals); i++ {
440 
+			t.Errorf("[%s] missing expected removal at position %d (%q)", tc.name, i, tc.expectedRemovals[i])
441 
+		}
442 
+		for i := len(tc.expectedRemovals); i < len(ms.removed); i++ {
443 
+			t.Errorf("[%s] removed unexpected metadata at position %d (%q)", tc.name, i, ms.removed[i])
444 
+		}
445 
+	}
446 
+}
447 
+
139 448 
 func taggedMetadata(key string, dgst string, sourceRepo string) metadata.V2Metadata {
140 449 
 	meta := metadata.V2Metadata{
141 450 
 		Digest:           digest.Digest(dgst),
... ... 
@@ -145,3 +461,114 @@ func taggedMetadata(key string, dgst string, sourceRepo string) metadata.V2Metad
145 145 
 	meta.HMAC = metadata.ComputeV2MetadataHMAC([]byte(key), &meta)
146 146 
 	return meta
147 147 
 }
148 
+
149 
+type mockRepo struct {
150 
+	t        *testing.T
151 
+	errors   map[digest.Digest]error
152 
+	blobs    map[digest.Digest]distribution.Descriptor
153 
+	requests []string
154 
+}
155 
+
156 
+var _ distribution.Repository = &mockRepo{}
157 
+
158 
+func (m *mockRepo) Named() distreference.Named {
159 
+	m.t.Fatalf("Named() not implemented")
160 
+	return nil
161 
+}
162 
+func (m *mockRepo) Manifests(ctc context.Context, options ...distribution.ManifestServiceOption) (distribution.ManifestService, error) {
163 
+	m.t.Fatalf("Manifests() not implemented")
164 
+	return nil, nil
165 
+}
166 
+func (m *mockRepo) Tags(ctc context.Context) distribution.TagService {
167 
+	m.t.Fatalf("Tags() not implemented")
168 
+	return nil
169 
+}
170 
+func (m *mockRepo) Blobs(ctx context.Context) distribution.BlobStore {
171 
+	return &mockBlobStore{
172 
+		repo: m,
173 
+	}
174 
+}
175 
+
176 
+type mockBlobStore struct {
177 
+	repo *mockRepo
178 
+}
179 
+
180 
+var _ distribution.BlobStore = &mockBlobStore{}
181 
+
182 
+func (m *mockBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) {
183 
+	m.repo.requests = append(m.repo.requests, dgst.String())
184 
+	if err, exists := m.repo.errors[dgst]; exists {
185 
+		return distribution.Descriptor{}, err
186 
+	}
187 
+	if desc, exists := m.repo.blobs[dgst]; exists {
188 
+		return desc, nil
189 
+	}
190 
+	return distribution.Descriptor{}, distribution.ErrBlobUnknown
191 
+}
192 
+func (m *mockBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) {
193 
+	m.repo.t.Fatal("Get() not implemented")
194 
+	return nil, nil
195 
+}
196 
+
197 
+func (m *mockBlobStore) Open(ctx context.Context, dgst digest.Digest) (distribution.ReadSeekCloser, error) {
198 
+	m.repo.t.Fatal("Open() not implemented")
199 
+	return nil, nil
200 
+}
201 
+
202 
+func (m *mockBlobStore) Put(ctx context.Context, mediaType string, p []byte) (distribution.Descriptor, error) {
203 
+	m.repo.t.Fatal("Put() not implemented")
204 
+	return distribution.Descriptor{}, nil
205 
+}
206 
+
207 
+func (m *mockBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
208 
+	m.repo.t.Fatal("Create() not implemented")
209 
+	return nil, nil
210 
+}
211 
+func (m *mockBlobStore) Resume(ctx context.Context, id string) (distribution.BlobWriter, error) {
212 
+	m.repo.t.Fatal("Resume() not implemented")
213 
+	return nil, nil
214 
+}
215 
+func (m *mockBlobStore) Delete(ctx context.Context, dgst digest.Digest) error {
216 
+	m.repo.t.Fatal("Delete() not implemented")
217 
+	return nil
218 
+}
219 
+func (m *mockBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error {
220 
+	m.repo.t.Fatalf("ServeBlob() not implemented")
221 
+	return nil
222 
+}
223 
+
224 
+type mockV2MetadataService struct {
225 
+	added   []metadata.V2Metadata
226 
+	removed []metadata.V2Metadata
227 
+}
228 
+
229 
+var _ metadata.V2MetadataService = &mockV2MetadataService{}
230 
+
231 
+func (*mockV2MetadataService) GetMetadata(diffID layer.DiffID) ([]metadata.V2Metadata, error) {
232 
+	return nil, nil
233 
+}
234 
+func (*mockV2MetadataService) GetDiffID(dgst digest.Digest) (layer.DiffID, error) {
235 
+	return "", nil
236 
+}
237 
+func (m *mockV2MetadataService) Add(diffID layer.DiffID, metadata metadata.V2Metadata) error {
238 
+	m.added = append(m.added, metadata)
239 
+	return nil
240 
+}
241 
+func (m *mockV2MetadataService) TagAndAdd(diffID layer.DiffID, hmacKey []byte, meta metadata.V2Metadata) error {
242 
+	meta.HMAC = metadata.ComputeV2MetadataHMAC(hmacKey, &meta)
243 
+	m.Add(diffID, meta)
244 
+	return nil
245 
+}
246 
+func (m *mockV2MetadataService) Remove(metadata metadata.V2Metadata) error {
247 
+	m.removed = append(m.removed, metadata)
248 
+	return nil
249 
+}
250 
+
251 
+type progressSink struct {
252 
+	t *testing.T
253 
+}
254 
+
255 
+func (s *progressSink) WriteProgress(p progress.Progress) error {
256 
+	s.t.Logf("progress update: %#+v", p)
257 
+	return nil
258 
+}