From ddc64d3ce9e3b7e6cf21ce42cf531dbb6ef83924 Mon Sep 17 00:00:00 2001 From: Lubos Mjachky Date: Mon, 17 Apr 2023 12:25:22 +0200 Subject: [PATCH] Sign manifests asynchronously closes #1208 --- CHANGES/1208.feature | 1 + pulp_container/app/tasks/sign.py | 75 ++++++++++++++++++++------------ 2 files changed, 49 insertions(+), 27 deletions(-) create mode 100644 CHANGES/1208.feature diff --git a/CHANGES/1208.feature b/CHANGES/1208.feature new file mode 100644 index 000000000..97bccda4a --- /dev/null +++ b/CHANGES/1208.feature @@ -0,0 +1 @@ +Started signing manifests asynchronously. This feature improves the performance of signing tasks. diff --git a/pulp_container/app/tasks/sign.py b/pulp_container/app/tasks/sign.py index 18b485dc9..4fa2a1ebc 100644 --- a/pulp_container/app/tasks/sign.py +++ b/pulp_container/app/tasks/sign.py @@ -1,8 +1,8 @@ +import asyncio import base64 import hashlib -import os -import tempfile +from aiofiles import tempfile from django.conf import settings from pulpcore.plugin.models import Repository @@ -19,6 +19,10 @@ SIGNATURE_TYPE, ) +SIGNING_TASKS_COUNTER = 10 + +semaphore = asyncio.Semaphore(SIGNING_TASKS_COUNTER) + def sign(repository_pk, signing_service_pk, reference, tags_list=None): """ @@ -27,8 +31,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 +51,40 @@ 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 + added_signatures = [] + 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 +99,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(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 +136,6 @@ def create_signature(manifest, reference, signing_service): data=encoded_sig, signed_manifest=manifest, ) - signature.save() + await signature.asave() return signature.pk