diff --git a/CHANGES/1208.feature b/CHANGES/1208.feature new file mode 100644 index 000000000..bd132fc14 --- /dev/null +++ b/CHANGES/1208.feature @@ -0,0 +1,3 @@ +Started signing manifests asynchronously. This feature improves the performance of signing tasks. +Additionally, setting ``MAX_PARALLEL_SIGNING_TASKS`` was introduced to cap the number of threads +used for parallel signing (defaults to ``10``). diff --git a/pulp_container/app/settings.py b/pulp_container/app/settings.py index f01a50938..583216689 100644 --- a/pulp_container/app/settings.py +++ b/pulp_container/app/settings.py @@ -48,3 +48,6 @@ } FLATPAK_INDEX = False + +# The number of allowed threads to sign manifests in parallel +MAX_PARALLEL_SIGNING_TASKS = 10 diff --git a/pulp_container/app/tasks/sign.py b/pulp_container/app/tasks/sign.py index 18b485dc9..38329419a 100644 --- a/pulp_container/app/tasks/sign.py +++ b/pulp_container/app/tasks/sign.py @@ -1,8 +1,9 @@ +import asyncio import base64 import hashlib -import os -import tempfile +from aiofiles import tempfile +from asgiref.sync import sync_to_async from django.conf import settings from pulpcore.plugin.models import Repository @@ -19,6 +20,8 @@ SIGNATURE_TYPE, ) +semaphore = asyncio.Semaphore(settings.MAX_PARALLEL_SIGNING_TASKS) + def sign(repository_pk, signing_service_pk, reference, tags_list=None): """ @@ -27,8 +30,8 @@ def sign(repository_pk, signing_service_pk, reference, tags_list=None): Create signature for each manifest that is specified and add it to the repo. If no manifests were specified, then sign all manifests in the repo. - What manifests to sign is identified by tag. - Manifest lists are not signed. Image manifests from manifest list are signed by digest. + What manifests to sign is identified by tags. + Manifest lists are signed too. Image manifests from the manifest lists are signed by tags. Args: repository_pk (uuid): A pk for a Repository for which a new Repository Version should be @@ -47,26 +50,39 @@ def sign(repository_pk, signing_service_pk, reference, tags_list=None): ) else: latest_repo_content_tags = latest_version.content.filter(pulp_type=Tag.get_pulp_type()) - latest_repo_tags = Tag.objects.filter(pk__in=latest_repo_content_tags) + latest_repo_tags = Tag.objects.filter(pk__in=latest_repo_content_tags).select_related( + "tagged_manifest" + ) signing_service = ManifestSigningService.objects.get(pk=signing_service_pk) - added_signatures = [] - for tag in latest_repo_tags: - tagged_manifest = tag.tagged_manifest - docker_reference = ":".join((reference, tag.name)) - signature_pk = create_signature(tagged_manifest, docker_reference, signing_service) - added_signatures.append(signature_pk) - if tagged_manifest.media_type in MANIFEST_MEDIA_TYPES.LIST: - # parse ML and sign per-arches - for manifest in tagged_manifest.listed_manifests.iterator(): - signature_pk = create_signature(manifest, docker_reference, signing_service) - added_signatures.append(signature_pk) + async def sign_manifests(): + added_signatures = [] + + async for tag in latest_repo_tags.aiterator(): + tagged_manifest = tag.tagged_manifest + docker_reference = ":".join((reference, tag.name)) + signature_pk = await create_signature( + tagged_manifest, docker_reference, signing_service + ) + added_signatures.append(signature_pk) + if tagged_manifest.media_type in MANIFEST_MEDIA_TYPES.LIST: + # parse ML and sign per-arches + manifests_iterator = tagged_manifest.listed_manifests.aiterator() + async for manifest in manifests_iterator: + signature_pk = await create_signature( + manifest, docker_reference, signing_service + ) + added_signatures.append(signature_pk) + + return added_signatures + + added_signatures = asyncio.run(sign_manifests()) added_signatures_qs = ManifestSignature.objects.filter(pk__in=added_signatures) with repository.new_version() as new_version: new_version.add_content(added_signatures_qs) -def create_signature(manifest, reference, signing_service): +async def create_signature(manifest, reference, signing_service): """ Create manifest signature. @@ -81,20 +97,23 @@ def create_signature(manifest, reference, signing_service): pk of created ManifestSignature. """ - with tempfile.TemporaryDirectory(dir=".") as working_directory: + async with semaphore: # download and write file for object storage + artifact = await manifest._artifacts.aget() if settings.DEFAULT_FILE_STORAGE != "pulpcore.app.models.storage.FileSystem": - manifest_file = tempfile.NamedTemporaryFile(dir=working_directory, delete=False) - artifact = manifest._artifacts.get() - manifest_file.write(artifact.file.read()) - manifest_file.flush() + async with tempfile.NamedTemporaryFile(dir=".", mode="wb", delete=False) as tf: + await tf.write(await sync_to_async(artifact.file.read)()) + await tf.flush() + artifact.file.close() - manifest_path = manifest_file.name + manifest_path = tf.name else: - manifest_path = manifest._artifacts.get().file.path - sig_path = os.path.join(working_directory, "signature") + manifest_path = artifact.file.path + + async with tempfile.NamedTemporaryFile(dir=".", prefix="signature") as tf: + sig_path = tf.name - signed = signing_service.sign( + signed = await signing_service.asign( manifest_path, env_vars={"REFERENCE": reference, "SIG_PATH": sig_path} ) @@ -115,6 +134,6 @@ def create_signature(manifest, reference, signing_service): data=encoded_sig, signed_manifest=manifest, ) - signature.save() + await signature.asave() return signature.pk