From 180d8847db1552ea55aee1394c775504361c63b9 Mon Sep 17 00:00:00 2001 From: Joseph Schorr Date: Mon, 26 Nov 2018 16:15:48 +0200 Subject: [PATCH] Fix image replication for images with remote layers --- data/registry_model/interface.py | 7 +-- data/registry_model/registry_oci_model.py | 10 ++++ data/registry_model/registry_pre_oci_model.py | 10 ++++ data/registry_model/shared.py | 26 +++++++++ data/registry_model/test/test_interface.py | 58 ++++++++++--------- endpoints/v2/manifest.py | 14 ++--- test/registry/registry_tests.py | 18 ++++++ 7 files changed, 103 insertions(+), 40 deletions(-) diff --git a/data/registry_model/interface.py b/data/registry_model/interface.py index febb17ad6..3e6ebe6c0 100644 --- a/data/registry_model/interface.py +++ b/data/registry_model/interface.py @@ -191,11 +191,8 @@ class RegistryDataInterface(object): """ Returns whether the given namespace exists and is enabled. """ @abstractmethod - def list_manifest_layers(self, manifest, include_placements=False): - """ Returns an *ordered list* of the layers found in the manifest, starting at the base and - working towards the leaf, including the associated Blob and its placements (if specified). - Returns None if the manifest could not be parsed and validated. - """ + def get_manifest_local_blobs(self, manifest, include_placements=False): + """ Returns the set of local blobs for the given manifest or None if none. """ @abstractmethod def list_parsed_manifest_layers(self, repository_ref, parsed_manifest, include_placements=False): diff --git a/data/registry_model/registry_oci_model.py b/data/registry_model/registry_oci_model.py index 237516edb..94916bf42 100644 --- a/data/registry_model/registry_oci_model.py +++ b/data/registry_model/registry_oci_model.py @@ -531,5 +531,15 @@ class OCIModel(SharedModel, RegistryDataInterface): return self._list_manifest_layers(repository_ref._db_id, parsed_manifest, include_placements, by_manifest=True) + def get_manifest_local_blobs(self, manifest, include_placements=False): + """ Returns the set of local blobs for the given manifest or None if none. """ + try: + manifest_row = database.Manifest.get(id=manifest._db_id) + except database.Manifest.DoesNotExist: + return None + + return self._get_manifest_local_blobs(manifest, manifest_row.repository_id, include_placements, + by_manifest=True) + oci_model = OCIModel() diff --git a/data/registry_model/registry_pre_oci_model.py b/data/registry_model/registry_pre_oci_model.py index 0ce9cdf40..0cb082318 100644 --- a/data/registry_model/registry_pre_oci_model.py +++ b/data/registry_model/registry_pre_oci_model.py @@ -584,5 +584,15 @@ class PreOCIModel(SharedModel, RegistryDataInterface): """ return self._list_manifest_layers(repository_ref._db_id, parsed_manifest, include_placements) + def get_manifest_local_blobs(self, manifest, include_placements=False): + """ Returns the set of local blobs for the given manifest or None if none. """ + try: + tag_manifest = database.TagManifest.get(id=manifest._db_id) + except database.TagManifest.DoesNotExist: + return None + + return self._get_manifest_local_blobs(manifest, tag_manifest.tag.repository_id, + include_placements) + pre_oci_model = PreOCIModel() diff --git a/data/registry_model/shared.py b/data/registry_model/shared.py index c220c9652..b4438bda7 100644 --- a/data/registry_model/shared.py +++ b/data/registry_model/shared.py @@ -290,6 +290,32 @@ class SharedModel: return LegacyImage.for_image(image, images_map=parent_images_map, blob=blob) + def _get_manifest_local_blobs(self, manifest, repo_id, include_placements=False, + by_manifest=False): + parsed = manifest.get_parsed_manifest() + if parsed is None: + return None + + local_blob_digests = list(set(parsed.local_blob_digests)) + if not len(local_blob_digests): + return [] + + blob_query = model.storage.lookup_repo_storages_by_content_checksum(repo_id, + local_blob_digests, + by_manifest=by_manifest) + blobs = [] + for image_storage in blob_query: + placements = None + if include_placements: + placements = list(model.storage.get_storage_locations(image_storage.uuid)) + + blob = Blob.for_image_storage(image_storage, + storage_path=model.storage.get_layer_path(image_storage), + placements=placements) + blobs.append(blob) + + return blobs + def _list_manifest_layers(self, repo_id, parsed, include_placements=False, by_manifest=False): """ Returns an *ordered list* of the layers found in the manifest, starting at the base and working towards the leaf, including the associated Blob and its placements (if specified). diff --git a/data/registry_model/test/test_interface.py b/data/registry_model/test/test_interface.py index 580dd2212..88cc65367 100644 --- a/data/registry_model/test/test_interface.py +++ b/data/registry_model/test/test_interface.py @@ -462,7 +462,7 @@ def test_is_namespace_enabled(namespace, expect_enabled, registry_model): ('devtable', 'history'), ('buynlarge', 'orgrepo'), ]) -def test_list_manifest_layers(repo_namespace, repo_name, registry_model): +def test_layers_and_blobs(repo_namespace, repo_name, registry_model): repository_ref = registry_model.lookup_repository(repo_namespace, repo_name) tags = registry_model.list_repository_tags(repository_ref) assert tags @@ -471,10 +471,14 @@ def test_list_manifest_layers(repo_namespace, repo_name, registry_model): manifest = registry_model.get_manifest_for_tag(tag) assert manifest - layers = registry_model.list_manifest_layers(manifest) + parsed = manifest.get_parsed_manifest() + assert parsed + + layers = registry_model.list_parsed_manifest_layers(repository_ref, parsed) assert layers - layers = registry_model.list_manifest_layers(manifest, include_placements=True) + layers = registry_model.list_parsed_manifest_layers(repository_ref, parsed, + include_placements=True) assert layers parsed_layers = list(manifest.get_parsed_manifest().layers) @@ -491,6 +495,9 @@ def test_list_manifest_layers(repo_namespace, repo_name, registry_model): assert manifest_layer.estimated_size(1) is not None + blobs = registry_model.get_manifest_local_blobs(manifest, include_placements=True) + assert {b.digest for b in blobs} == set(parsed.local_blob_digests) + def test_manifest_remote_layers(oci_model): # Create a config blob for testing. @@ -524,7 +531,8 @@ def test_manifest_remote_layers(oci_model): 'sometag', storage) assert created_manifest - layers = oci_model.list_manifest_layers(created_manifest) + layers = oci_model.list_parsed_manifest_layers(repository_ref, + created_manifest.get_parsed_manifest()) assert len(layers) == 1 assert layers[0].layer_info.is_remote assert layers[0].layer_info.urls == ['http://hello/world'] @@ -601,25 +609,25 @@ def test_torrent_info(registry_model): tag = registry_model.get_repo_tag(repository_ref, 'latest') manifest = registry_model.get_manifest_for_tag(tag) - layers = registry_model.list_manifest_layers(manifest) - assert layers + blobs = registry_model.get_manifest_local_blobs(manifest) + assert blobs - assert registry_model.get_torrent_info(layers[0].blob) is None - registry_model.set_torrent_info(layers[0].blob, 2, 'foo') + assert registry_model.get_torrent_info(blobs[0]) is None + registry_model.set_torrent_info(blobs[0], 2, 'foo') # Set it again exactly, which should be a no-op. - registry_model.set_torrent_info(layers[0].blob, 2, 'foo') + registry_model.set_torrent_info(blobs[0], 2, 'foo') # Check the information we've set. - torrent_info = registry_model.get_torrent_info(layers[0].blob) + torrent_info = registry_model.get_torrent_info(blobs[0]) assert torrent_info is not None assert torrent_info.piece_length == 2 assert torrent_info.pieces == 'foo' # Try setting it again. Nothing should happen. - registry_model.set_torrent_info(layers[0].blob, 3, 'bar') + registry_model.set_torrent_info(blobs[0], 3, 'bar') - torrent_info = registry_model.get_torrent_info(layers[0].blob) + torrent_info = registry_model.get_torrent_info(blobs[0]) assert torrent_info is not None assert torrent_info.piece_length == 2 assert torrent_info.pieces == 'foo' @@ -680,19 +688,19 @@ def test_mount_blob_into_repository(pre_oci_model): target_repository_ref = pre_oci_model.lookup_repository('devtable', 'complex') - layers = pre_oci_model.list_manifest_layers(manifest, include_placements=True) - assert layers + blobs = pre_oci_model.get_manifest_local_blobs(manifest, include_placements=True) + assert blobs - for layer in layers: + for blob in blobs: # Ensure the blob doesn't exist under the repository. - assert not pre_oci_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest) + assert not pre_oci_model.get_repo_blob_by_digest(target_repository_ref, blob.digest) # Mount the blob into the repository. - assert pre_oci_model.mount_blob_into_repository(layer.blob, target_repository_ref, 60) + assert pre_oci_model.mount_blob_into_repository(blob, target_repository_ref, 60) # Ensure it now exists. - found = pre_oci_model.get_repo_blob_by_digest(target_repository_ref, layer.blob.digest) - assert found == layer.blob + found = pre_oci_model.get_repo_blob_by_digest(target_repository_ref, blob.digest) + assert found == blob class SomeException(Exception): @@ -706,10 +714,10 @@ def test_get_cached_repo_blob(registry_model): latest_tag = registry_model.get_repo_tag(repository_ref, 'latest') manifest = registry_model.get_manifest_for_tag(latest_tag) - layers = registry_model.list_manifest_layers(manifest, include_placements=True) - assert layers + blobs = registry_model.get_manifest_local_blobs(manifest, include_placements=True) + assert blobs - blob = layers[0].blob + blob = blobs[0] # Load a blob to add it to the cache. found = registry_model.get_cached_repo_blob(model_cache, 'devtable', 'simple', blob.digest) @@ -764,9 +772,6 @@ def test_create_manifest_and_retarget_tag(registry_model): assert tag.name == 'anothertag' assert another_manifest.get_parsed_manifest().manifest_dict == sample_manifest.manifest_dict - layers = registry_model.list_manifest_layers(another_manifest) - assert len(layers) == 1 - def test_get_schema1_parsed_manifest(registry_model): repository_ref = registry_model.lookup_repository('devtable', 'simple') @@ -804,8 +809,5 @@ def test_create_manifest_and_retarget_tag_with_labels(registry_model): assert tag.name == 'anothertag' assert another_manifest.get_parsed_manifest().manifest_dict == sample_manifest.manifest_dict - layers = registry_model.list_manifest_layers(another_manifest) - assert len(layers) == 1 - # Ensure the labels were applied. assert tag.lifetime_end_ms is not None diff --git a/endpoints/v2/manifest.py b/endpoints/v2/manifest.py index c8a81ea6f..bba45e062 100644 --- a/endpoints/v2/manifest.py +++ b/endpoints/v2/manifest.py @@ -250,13 +250,13 @@ def _write_manifest_and_log(namespace_name, repo_name, tag_name, manifest_impl): # Queue all blob manifests for replication. if features.STORAGE_REPLICATION: - layers = registry_model.list_manifest_layers(manifest) - if layers is None: - raise ManifestInvalid() - - with queue_replication_batch(namespace_name) as queue_storage_replication: - for layer in layers: - queue_storage_replication(layer.blob) + blobs = registry_model.get_manifest_local_blobs(manifest) + if blobs is None: + logger.error('Could not lookup blobs for manifest `%s`', manifest.digest) + else: + with queue_replication_batch(namespace_name) as queue_storage_replication: + for blob_digest in blobs: + queue_storage_replication(blob_digest) track_and_log('push_repo', repository_ref, tag=tag_name) spawn_notification(repository_ref, 'repo_push', {'updated_tags': [tag_name]}) diff --git a/test/registry/registry_tests.py b/test/registry/registry_tests.py index e31836c51..2f75ff845 100644 --- a/test/registry/registry_tests.py +++ b/test/registry/registry_tests.py @@ -459,6 +459,24 @@ def test_image_replication(pusher, puller, basic_images, liveserver_session, app assert r.text == 'OK' +def test_image_replication_empty_layers(pusher, puller, images_with_empty_layer, liveserver_session, + app_reloader, liveserver, registry_server_executor): + """ Test: Ensure that entries are created for replication of the images pushed. """ + credentials = ('devtable', 'password') + + with FeatureFlagValue('STORAGE_REPLICATION', True, registry_server_executor.on(liveserver)): + pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', images_with_empty_layer, + credentials=credentials) + + result = puller.pull(liveserver_session, 'devtable', 'newrepo', 'latest', + images_with_empty_layer, credentials=credentials) + + # Ensure that entries were created for each image. + for image_id in result.image_ids.values(): + r = registry_server_executor.on(liveserver).get_storage_replication_entry(image_id) + assert r.text == 'OK' + + @pytest.mark.parametrize('repo_name, expected_failure', [ ('something', None), ('some/slash', Failures.SLASH_REPOSITORY),