Merge ~pappacena/launchpad:oci-multi-arch-upload into launchpad:master

[Thiago F. Pappacena]

Upload multi-arch OCI images

6c75d70... by Thiago F. Pappacena on 2020-10-09

Merge branch 'master' into oci-multi-arch-upload

[Colin Watson (cjwatson)]

I can't effectively review the actual manifest handling - it's probably best to ask Tom to check that over.

review: Needs Fixing

2870200... by Thiago F. Pappacena on 2020-10-09

Code style, typo and better code documentation

[Thiago F. Pappacena (pappacena)]

Replied to the comments about `allBuildsUploaded` method. I would like your opinion on that while I work on the job retry details.

08c1c68... by Thiago F. Pappacena on 2020-10-09

Refactoring and adding tests for concurrent oci img upload jobs check

[Thiago F. Pappacena (pappacena)]

Added retry strategy and some tests to ensure that we are using database locks the way we expect, to queue up the check for manifest list upload.

4bad06d... by Thiago F. Pappacena on 2020-10-09

Merge branch 'master' into oci-multi-arch-upload

8b6e704... by Thiago F. Pappacena on 2020-10-09

Fixing exception catching in test

15622d5... by Thiago F. Pappacena on 2020-10-09

Fixing _upload_layer mock return value in tests

[Colin Watson (cjwatson)]

review: Approve

bd79d6e... by Thiago F. Pappacena on 2020-10-15

Checking also final result on concurrent OCIRecipeUploadJob

caa19b0... by Thiago F. Pappacena on 2020-10-15

Refactoring

[Thiago F. Pappacena (pappacena)]

Addressed all comments. As suggested by twom, I'll test it with microk8s before merging it.

[Thiago F. Pappacena (pappacena)]

Tested with microk8s and it seems to be ok.

diff --git a/database/schema/security.cfg b/database/schema/security.cfg
index 8a9a31b..d5a23f1 100644
--- a/database/schema/security.cfg
+++ b/database/schema/security.cfg
@@ -2710,6 +2710,7 @@ public.account                          = SELECT
 public.archive                          = SELECT
 public.branch                           = SELECT
 public.builder                          = SELECT
+public.builderprocessor                 = SELECT
 public.buildfarmjob                     = SELECT, INSERT
 public.buildqueue                       = SELECT, INSERT, UPDATE
 public.distribution                     = SELECT
diff --git a/lib/lp/oci/interfaces/ocirecipe.py b/lib/lp/oci/interfaces/ocirecipe.py
index 0bd3cd2..131c4d3 100644
--- a/lib/lp/oci/interfaces/ocirecipe.py
+++ b/lib/lp/oci/interfaces/ocirecipe.py
@@ -178,6 +178,16 @@ class IOCIRecipeBuildRequest(Interface):
         title=_("If set, limit builds to these architecture tags."),
         value_type=TextLine(), required=False, readonly=True)
 
+    uploaded_manifests = Dict(
+        title=_("A dict of manifest information per build."),
+        key_type=Int(), value_type=Dict(),
+        required=False, readonly=True)
+
+    def addUploadedManifest(build_id, manifest_info):
+        """Add the manifest information for one of the builds in this
+        BuildRequest.
+        """
+
 
 class IOCIRecipeView(Interface):
     """`IOCIRecipe` attributes that require launchpad.View permission."""
diff --git a/lib/lp/oci/interfaces/ocirecipejob.py b/lib/lp/oci/interfaces/ocirecipejob.py
index 104a2ff..2b59faf 100644
--- a/lib/lp/oci/interfaces/ocirecipejob.py
+++ b/lib/lp/oci/interfaces/ocirecipejob.py
@@ -19,6 +19,8 @@ from zope.interface import (
     )
 from zope.schema import (
     Datetime,
+    Dict,
+    Int,
     List,
     TextLine,
     )
@@ -78,6 +80,16 @@ class IOCIRecipeRequestBuildsJob(IRunnableJob):
     error_message = TextLine(
         title=_("Error message"), required=False, readonly=True)
 
+    uploaded_manifests = Dict(
+        title=_("A dict of manifest information per build."),
+        key_type=Int(), value_type=Dict(),
+        required=False, readonly=True)
+
+    def addUploadedManifest(build_id, manifest_info):
+        """Add the manifest information for one of the builds in this
+        BuildRequest.
+        """
+
 
 class IOCIRecipeRequestBuildsJobSource(IJobSource):
 
diff --git a/lib/lp/oci/interfaces/ociregistryclient.py b/lib/lp/oci/interfaces/ociregistryclient.py
index 184a93f..71af23a 100644
--- a/lib/lp/oci/interfaces/ociregistryclient.py
+++ b/lib/lp/oci/interfaces/ociregistryclient.py
@@ -54,3 +54,6 @@ class IOCIRegistryClient(Interface):
 
         :param build: The `IOCIRecipeBuild` to upload.
         """
+
+    def uploadManifestList(build_request):
+        """Upload the "fat manifest" which aggregates all platforms built."""
diff --git a/lib/lp/oci/model/ocirecipe.py b/lib/lp/oci/model/ocirecipe.py
index fd21dde..ab7f655 100644
--- a/lib/lp/oci/model/ocirecipe.py
+++ b/lib/lp/oci/model/ocirecipe.py
@@ -38,7 +38,10 @@ from zope.component import (
 from zope.event import notify
 from zope.interface import implementer
 from zope.security.interfaces import Unauthorized
-from zope.security.proxy import removeSecurityProxy
+from zope.security.proxy import (
+    isinstance as zope_isinstance,
+    removeSecurityProxy,
+    )
 
 from lp.app.interfaces.launchpad import ILaunchpadCelebrities
 from lp.app.interfaces.security import IAuthorization
@@ -697,6 +700,13 @@ class OCIRecipeBuildRequest:
         return self.job.date_finished
 
     @property
+    def uploaded_manifests(self):
+        return self.job.uploaded_manifests
+
+    def addUploadedManifest(self, build_id, manifest_info):
+        self.job.addUploadedManifest(build_id, manifest_info)
+
+    @property
     def status(self):
         status_map = {
             JobStatus.WAITING: OCIRecipeBuildRequestStatus.PENDING,
@@ -714,3 +724,8 @@ class OCIRecipeBuildRequest:
     @property
     def builds(self):
         return self.job.builds
+
+    def __eq__(self, other):
+        if not zope_isinstance(other, self.__class__):
+            return False
+        return self.id == other.id
diff --git a/lib/lp/oci/model/ocirecipebuildjob.py b/lib/lp/oci/model/ocirecipebuildjob.py
index 73fa4d6..991b9a4 100644
--- a/lib/lp/oci/model/ocirecipebuildjob.py
+++ b/lib/lp/oci/model/ocirecipebuildjob.py
@@ -11,6 +11,8 @@ __all__ = [
     'OCIRecipeBuildJobType',
     ]
 
+from datetime import timedelta
+import random
 
 from lazr.delegates import delegate_to
 from lazr.enum import (
@@ -41,8 +43,12 @@ from lp.oci.interfaces.ociregistryclient import (
     OCIRegistryError,
     )
 from lp.services.database.enumcol import DBEnum
-from lp.services.database.interfaces import IStore
+from lp.services.database.interfaces import (
+    IMasterStore,
+    IStore,
+    )
 from lp.services.database.stormbase import StormBase
+from lp.services.job.interfaces.job import JobStatus
 from lp.services.job.model.job import (
     EnumeratedSubclass,
     Job,
@@ -156,16 +162,34 @@ class OCIRecipeBuildJobDerived(
 @implementer(IOCIRegistryUploadJob)
 @provider(IOCIRegistryUploadJobSource)
 class OCIRegistryUploadJob(OCIRecipeBuildJobDerived):
+    """Manages the OCI image upload to registries.
+
+    This job coordinates with other OCIRegistryUploadJob in a way that the
+    last job uploading its layers and manifest will also upload the
+    manifest list with all the previously built OCI images uploaded from
+    other architectures in the same build request. To avoid race conditions,
+    we synchronize that using a SELECT ... FOR UPDATE at database level to
+    make sure that the status is consistent across all the upload jobs.
+    """
 
     class_job_type = OCIRecipeBuildJobType.REGISTRY_UPLOAD
 
+    class ManifestListUploadError(Exception):
+        pass
+
+    retry_error_types = (ManifestListUploadError, )
+    max_retries = 5
+
     @classmethod
     def create(cls, build):
         """See `IOCIRegistryUploadJobSource`"""
         edited_fields = set()
         with notify_modified(build, edited_fields) as before_modification:
+            json_data = {
+                "build_uploaded": False,
+            }
             oci_build_job = OCIRecipeBuildJob(
-                build, cls.class_job_type, {})
+                build, cls.class_job_type, json_data)
             job = cls(oci_build_job)
             job.celeryRunOnCommit()
             del get_property_cache(build).last_registry_upload_job
@@ -174,6 +198,14 @@ class OCIRegistryUploadJob(OCIRecipeBuildJobDerived):
                 edited_fields.add("registry_upload_status")
         return job
 
+    @property
+    def retry_delay(self):
+        dithering_secs = int(random.random() * 60)
+        # Adds some random seconds between 0 and 60 to minimize the
+        # likelihood of synchronized retries holding locks on the database
+        # at the same time.
+        return timedelta(minutes=10, seconds=dithering_secs)
+
     # Ideally we'd just override Job._set_status or similar, but
     # lazr.delegates makes that difficult, so we use this to override all
     # the individual Job lifecycle methods instead.
@@ -227,6 +259,74 @@ class OCIRegistryUploadJob(OCIRecipeBuildJobDerived):
         """See `IOCIRegistryUploadJob`."""
         self.json_data["errors"] = errors
 
+    @property
+    def build_uploaded(self):
+        return self.json_data.get("build_uploaded", False)
+
+    @build_uploaded.setter
+    def build_uploaded(self, value):
+        self.json_data["build_uploaded"] = bool(value)
+
+    def allBuildsUploaded(self, build_request):
+        """Returns True if all builds of the given build_request already
+        finished uploading. False otherwise.
+
+        Note that this method locks all upload jobs at database level,
+        preventing them from updating their status until the end of the
+        current transaction. Use it with caution.
+        """
+        builds = list(build_request.builds)
+        uploads_per_build = {i: list(i.registry_upload_jobs) for i in builds}
+        upload_jobs = sum(uploads_per_build.values(), [])
+
+        # Lock the Job rows, so no other job updates its status until the
+        # end of this job's transaction. This is done to avoid race conditions,
+        # where 2 upload jobs could be running simultaneously and none of them
+        # realises that is the last upload.
+        # Note also that new upload jobs might be created between the
+        # transaction begin and this lock takes place, but in this case the
+        # new upload is either a retry from a failed upload, or the first
+        # upload for one of the existing builds. Either way, we would see that
+        # build as "not uploaded yet", which is ok for this method, and the
+        # new job will block until these locks are released, so we should be
+        # safe.
+        store = IMasterStore(builds[0])
+        placeholders = ', '.join('?' for _ in upload_jobs)
+        sql = (
+            "SELECT id, status FROM job WHERE id IN (%s) FOR UPDATE"
+            % placeholders)
+        job_status = {
+            job_id: JobStatus.items[status] for job_id, status in
+            store.execute(sql, [i.job_id for i in upload_jobs])}
+
+        for build, upload_jobs in uploads_per_build.items():
+            has_finished_upload = any(
+                job_status[i.job_id] == JobStatus.COMPLETED
+                or i.job_id == self.job_id
+                for i in upload_jobs)
+            if not has_finished_upload:
+                return False
+        return True
+
+    def uploadManifestList(self, client):
+        """Uploads the aggregated manifest list for all builds in the
+        current build request.
+        """
+        # The "allBuildsUploaded" call will lock, on the database,
+        # all upload jobs for update until this transaction finishes.
+        # So, make sure this is the last thing being done by this job.
+        build_request = self.build.build_request
+        if not build_request:
+            return
+        try:
+            if self.allBuildsUploaded(build_request):
+                client.uploadManifestList(build_request)
+        except OCIRegistryError:
+            # Do not retry automatically on known OCI registry errors.
+            raise
+        except Exception as e:
+            raise self.ManifestListUploadError(str(e))
+
     def run(self):
         """See `IRunnableJob`."""
         client = getUtility(IOCIRegistryClient)
@@ -234,7 +334,21 @@ class OCIRegistryUploadJob(OCIRecipeBuildJobDerived):
         # it will need to gain retry support.
         try:
             try:
-                client.upload(self.build)
+                if not self.build_uploaded:
+                    client.upload(self.build)
+                    self.build_uploaded = True
+
+                self.uploadManifestList(client)
+                # Force this job status to COMPLETED in the same transaction we
+                # called `allBuildsUpdated` (in the uploadManifestList call
+                # above) to release the lock already including the new status.
+                # This way, any other transaction that was blocked waiting to
+                # get the info about the upload jobs will immediately have the
+                # new status of this job, avoiding race conditions. Keep the
+                # `manage_transaction=False` to prevent the method from
+                # commiting at the wrong moment.
+                self.complete(manage_transaction=False)
+
             except OCIRegistryError as e:
                 self.error_summary = str(e)
                 self.errors = e.errors
diff --git a/lib/lp/oci/model/ocirecipejob.py b/lib/lp/oci/model/ocirecipejob.py
index 15eac4f..3320cff 100644
--- a/lib/lp/oci/model/ocirecipejob.py
+++ b/lib/lp/oci/model/ocirecipejob.py
@@ -169,6 +169,8 @@ class OCIRecipeRequestBuildsJob(OCIRecipeJobDerived):
             "requester": requester.id,
             "architectures": (
                 list(architectures) if architectures is not None else None),
+            # A dict of build_id: manifest location
+            "uploaded_manifests": {}
         }
         recipe_job = OCIRecipeJob(recipe, cls.class_job_type, metadata)
         job = cls(recipe_job)
@@ -258,6 +260,16 @@ class OCIRecipeRequestBuildsJob(OCIRecipeJobDerived):
         architectures = self.metadata["architectures"]
         return set(architectures) if architectures is not None else None
 
+    @property
+    def uploaded_manifests(self):
+        return {
+            # Converts keys to integer since saving json to database
+            # converts them to strings.
+            int(k): v for k, v in self.metadata["uploaded_manifests"].items()}
+
+    def addUploadedManifest(self, build_id, manifest_info):
+        self.metadata["uploaded_manifests"][int(build_id)] = manifest_info
+
     def run(self):
         """See `IRunnableJob`."""
         requester = self.requester
diff --git a/lib/lp/oci/model/ociregistryclient.py b/lib/lp/oci/model/ociregistryclient.py
index 7561267..7c67b57 100644
--- a/lib/lp/oci/model/ociregistryclient.py
+++ b/lib/lp/oci/model/ociregistryclient.py
@@ -229,17 +229,59 @@ class OCIRegistryClient:
         """
         # XXX twom 2020-04-17 This needs to include OCIProjectSeries and
         # base image name
-
         return "{}".format("edge")
 
     @classmethod
+    def _uploadRegistryManifest(cls, http_client, registry_manifest,
+                                push_rule, build=None):
+        """Uploads the build manifest, returning its content information.
+
+        The returned information can be used to create a Manifest list
+        including the uploaded manifest, for example, in order to create
+        multi-architecture images.
+
+        :return: A dict with {"digest": "sha256:xxx", "size": total_bytes}
+        """
+        digest = None
+        data = json.dumps(registry_manifest)
+        size = len(data)
+        content_type = registry_manifest.get(
+            "mediaType",
+            "application/vnd.docker.distribution.manifest.v2+json")
+        if build is None:
+            # When uploading a manifest list, use the tag.
+            tag = cls._calculateTag(build, push_rule)
+        else:
+            # When uploading individual build manifests, use their digest.
+            tag = "sha256:%s" % hashlib.sha256(data).hexdigest()
+        try:
+            manifest_response = http_client.requestPath(
+                "/manifests/{}".format(tag),
+                data=data,
+                headers={"Content-Type": content_type},
+                method="PUT")
+            digest = manifest_response.headers.get("Docker-Content-Digest")
+        except HTTPError as http_error:
+            manifest_response = http_error.response
+        if manifest_response.status_code != 201:
+            if build:
+                msg = "Failed to upload manifest for {} ({}) in {}".format(
+                    build.recipe.name, push_rule.registry_url, build.id)
+            else:
+                msg = ("Failed to upload manifest of manifests for"
+                       " {} ({})").format(
+                    push_rule.recipe.name, push_rule.registry_url)
+            raise cls._makeRegistryError(
+                ManifestUploadFailed, msg, manifest_response)
+        return {"digest": digest, "size": size}
+
+    @classmethod
     def _upload_to_push_rule(
             cls, push_rule, build, manifest, digests, preloaded_data):
         http_client = RegistryHTTPClient.getInstance(push_rule)
 
         for section in manifest:
-            # Work out names and tags
-            tag = cls._calculateTag(build, push_rule)
+            # Work out names
             file_data = preloaded_data[section["Config"]]
             config = file_data["config_file"]
             #  Upload the layers involved
@@ -269,24 +311,13 @@ class OCIRegistryClient:
                 layer_sizes)
 
             # Upload the registry manifest
-            try:
-                manifest_response = http_client.requestPath(
-                    "/manifests/{}".format(tag),
-                    json=registry_manifest,
-                    headers={
-                        "Content-Type":
-                            "application/"
-                            "vnd.docker.distribution.manifest.v2+json"
-                    },
-                    method="PUT")
-            except HTTPError as http_error:
-                manifest_response = http_error.response
-            if manifest_response.status_code != 201:
-                raise cls._makeRegistryError(
-                    ManifestUploadFailed,
-                    "Failed to upload manifest for {} ({}) in {}".format(
-                        build.recipe.name, push_rule.registry_url, build.id),
-                    manifest_response)
+            manifest = cls._uploadRegistryManifest(
+                http_client, registry_manifest, push_rule, build)
+
+            # Save the uploaded manifest location, so we can use it in case
+            # this is a multi-arch image upload.
+            if build.build_request:
+                build.build_request.addUploadedManifest(build.id, manifest)
 
     @classmethod
     def upload(cls, build):
@@ -318,6 +349,43 @@ class OCIRegistryClient:
         elif len(exceptions) > 1:
             raise MultipleOCIRegistryError(exceptions)
 
+    @classmethod
+    def makeMultiArchManifest(cls, build_request):
+        """Returns the multi-arch manifest content including all builds of
+        the given build_request.
+        """
+        manifests = []
+        for build in build_request.builds:
+            build_manifest = build_request.uploaded_manifests.get(build.id)
+            if not build_manifest:
+                continue
+            digest = build_manifest["digest"]
+            size = build_manifest["size"]
+            arch = build.processor.name
+            manifests.append({
+                "mediaType": ("application/"
+                              "vnd.docker.distribution.manifest.v2+json"),
+                "size": size,
+                "digest": digest,
+                "platform": {"architecture": arch, "os": "linux"}
+            })
+        return {
+          "schemaVersion": 2,
+          "mediaType": ("application/"
+                        "vnd.docker.distribution.manifest.list.v2+json"),
+          "manifests": manifests}
+
+    @classmethod
+    def uploadManifestList(cls, build_request):
+        """Uploads to all build_request.recipe.push_rules the manifest list
+        for the builds in the given build_request.
+        """
+        multi_manifest_content = cls.makeMultiArchManifest(build_request)
+        for push_rule in build_request.recipe.push_rules:
+            http_client = RegistryHTTPClient.getInstance(push_rule)
+            cls._uploadRegistryManifest(
+                http_client, multi_manifest_content, push_rule, build=None)
+
 
 class OCIRegistryAuthenticationError(Exception):
     def __init__(self, msg, http_error=None):
@@ -436,8 +504,8 @@ class BearerTokenRegistryClient(RegistryHTTPClient):
 
         :param auth_retry: Should we authenticate and retry the request if
                            it fails with HTTP 401 code?"""
+        headers = request_kwargs.pop("headers", {})
         try:
-            headers = request_kwargs.pop("headers", {})
             if self.auth_token is not None:
                 headers["Authorization"] = "Bearer %s" % self.auth_token
             return proxy_urlfetch(url, headers=headers, **request_kwargs)
@@ -445,5 +513,6 @@ class BearerTokenRegistryClient(RegistryHTTPClient):
             if auth_retry and e.response.status_code == 401:
                 self.authenticate(e.response)
                 return self.request(
-                    url, auth_retry=False, *args, **request_kwargs)
+                    url, auth_retry=False, headers=headers,
+                    *args, **request_kwargs)
             raise
diff --git a/lib/lp/oci/tests/test_ocirecipebuildjob.py b/lib/lp/oci/tests/test_ocirecipebuildjob.py
index fd3e0ff..7b66c1d 100644
--- a/lib/lp/oci/tests/test_ocirecipebuildjob.py
+++ b/lib/lp/oci/tests/test_ocirecipebuildjob.py
@@ -7,6 +7,13 @@ from __future__ import absolute_import, print_function, unicode_literals
 
 __metaclass__ = type
 
+from datetime import (
+    datetime,
+    timedelta,
+    )
+import threading
+import time
+
 from fixtures import FakeLogger
 from testtools.matchers import (
     Equals,
@@ -14,11 +21,15 @@ from testtools.matchers import (
     MatchesDict,
     MatchesListwise,
     MatchesStructure,
+    Not,
     )
 import transaction
+from zope.component import getUtility
 from zope.interface import implementer
+from zope.security.proxy import removeSecurityProxy
 
 from lp.buildmaster.enums import BuildStatus
+from lp.buildmaster.interfaces.processor import IProcessorSet
 from lp.oci.interfaces.ocirecipe import (
     OCI_RECIPE_ALLOW_CREATE,
     OCI_RECIPE_WEBHOOKS_FEATURE_FLAG,
@@ -27,6 +38,7 @@ from lp.oci.interfaces.ocirecipebuildjob import (
     IOCIRecipeBuildJob,
     IOCIRegistryUploadJob,
     )
+from lp.oci.interfaces.ocirecipejob import IOCIRecipeRequestBuildsJobSource
 from lp.oci.interfaces.ociregistryclient import (
     BlobUploadFailed,
     IOCIRegistryClient,
@@ -37,12 +49,17 @@ from lp.oci.model.ocirecipebuildjob import (
     OCIRecipeBuildJobType,
     OCIRegistryUploadJob,
     )
+from lp.services.compat import mock
 from lp.services.config import config
 from lp.services.features.testing import FeatureFixture
+from lp.services.job.interfaces.job import JobStatus
 from lp.services.job.runner import JobRunner
 from lp.services.webapp import canonical_url
 from lp.services.webhooks.testing import LogsScheduledWebhooks
-from lp.testing import TestCaseWithFactory
+from lp.testing import (
+    admin_logged_in,
+    TestCaseWithFactory,
+    )
 from lp.testing.dbuser import dbuser
 from lp.testing.fakemethod import FakeMethod
 from lp.testing.fixture import ZopeUtilityFixture
@@ -69,6 +86,7 @@ class FakeRegistryClient:
 
     def __init__(self):
         self.upload = FakeMethod()
+        self.uploadManifestList = FakeMethod()
 
 
 class FakeOCIBuildJob(OCIRecipeBuildJobDerived):
@@ -122,22 +140,28 @@ class TestOCIRegistryUploadJob(TestCaseWithFactory):
         ocibuild = self.factory.makeOCIRecipeBuild(
             builder=self.factory.makeBuilder(), **kwargs)
         ocibuild.updateStatus(BuildStatus.FULLYBUILT)
-        self.factory.makeWebhook(
-            target=ocibuild.recipe, event_types=["oci-recipe:build:0.1"])
+        self.makeWebhook(ocibuild.recipe)
         return ocibuild
 
+    def makeWebhook(self, recipe):
+        self.factory.makeWebhook(
+            target=recipe, event_types=["oci-recipe:build:0.1"])
+
     def assertWebhookDeliveries(self, ocibuild,
                                 expected_registry_upload_statuses, logger):
         hook = ocibuild.recipe.webhooks.one()
         deliveries = list(hook.deliveries)
         deliveries.reverse()
+        build_req_url = (
+            None if ocibuild.build_request is None
+            else canonical_url(ocibuild.build_request, force_local_path=True))
         expected_payloads = [{
             "recipe_build": Equals(
                 canonical_url(ocibuild, force_local_path=True)),
             "action": Equals("status-changed"),
             "recipe": Equals(
                 canonical_url(ocibuild.recipe, force_local_path=True)),
-            "build_request": Is(None),
+            "build_request": Equals(build_req_url),
             "status": Equals("Successfully built"),
             "registry_upload_status": Equals(expected),
             } for expected in expected_registry_upload_statuses]
@@ -165,22 +189,242 @@ class TestOCIRegistryUploadJob(TestCaseWithFactory):
         job = OCIRegistryUploadJob.create(ocibuild)
         self.assertProvides(job, IOCIRegistryUploadJob)
 
+    def makeRecipe(self, include_i386=True, include_amd64=True):
+        i386 = getUtility(IProcessorSet).getByName("386")
+        amd64 = getUtility(IProcessorSet).getByName("amd64")
+        recipe = self.factory.makeOCIRecipe()
+        distroseries = self.factory.makeDistroSeries(
+            distribution=recipe.oci_project.distribution)
+        distro_i386 = self.factory.makeDistroArchSeries(
+            distroseries=distroseries, architecturetag="i386",
+            processor=i386)
+        distro_i386.addOrUpdateChroot(self.factory.makeLibraryFileAlias())
+        distro_amd64 = self.factory.makeDistroArchSeries(
+            distroseries=distroseries, architecturetag="amd64",
+            processor=amd64)
+        distro_amd64.addOrUpdateChroot(self.factory.makeLibraryFileAlias())
+
+        archs = []
+        if include_i386:
+            archs.append(i386)
+        if include_amd64:
+            archs.append(amd64)
+        recipe.setProcessors(archs)
+        return recipe
+
+    def makeBuildRequest(self, include_i386=True, include_amd64=True):
+        recipe = self.makeRecipe(include_i386, include_amd64)
+        # Creates a build request with a build in it.
+        build_request = recipe.requestBuilds(recipe.owner)
+        with admin_logged_in():
+            jobs = getUtility(IOCIRecipeRequestBuildsJobSource).iterReady()
+            with dbuser(config.IOCIRecipeRequestBuildsJobSource.dbuser):
+                JobRunner(jobs).runAll()
+        return build_request
+
     def test_run(self):
         logger = self.useFixture(FakeLogger())
-        ocibuild = self.makeOCIRecipeBuild()
+        build_request = self.makeBuildRequest(include_i386=False)
+        recipe = build_request.recipe
+
+        self.assertEqual(1, build_request.builds.count())
+        ocibuild = build_request.builds[0]
+        ocibuild.updateStatus(BuildStatus.FULLYBUILT)
+        self.makeWebhook(recipe)
+
         self.assertContentEqual([], ocibuild.registry_upload_jobs)
         job = OCIRegistryUploadJob.create(ocibuild)
         client = FakeRegistryClient()
         self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
         with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
             run_isolated_jobs([job])
+
         self.assertEqual([((ocibuild,), {})], client.upload.calls)
+        self.assertEqual([((build_request,), {})],
+                         client.uploadManifestList.calls)
         self.assertContentEqual([job], ocibuild.registry_upload_jobs)
         self.assertIsNone(job.error_summary)
         self.assertIsNone(job.errors)
         self.assertEqual([], pop_notifications())
         self.assertWebhookDeliveries(ocibuild, ["Pending", "Uploaded"], logger)
 
+    def test_run_multiple_architectures(self):
+        build_request = self.makeBuildRequest()
+        builds = build_request.builds
+        self.assertEqual(2, builds.count())
+        self.assertEqual(builds[0].build_request, builds[1].build_request)
+
+        upload_jobs = []
+        for build in builds:
+            self.assertContentEqual([], build.registry_upload_jobs)
+            upload_jobs.append(OCIRegistryUploadJob.create(build))
+
+        client = FakeRegistryClient()
+        self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
+
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            JobRunner([upload_jobs[0]]).runAll()
+        self.assertEqual([((builds[0],), {})], client.upload.calls)
+        self.assertEqual([], client.uploadManifestList.calls)
+
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            JobRunner([upload_jobs[1]]).runAll()
+        self.assertEqual(
+            [((builds[0],), {}), ((builds[1],), {})], client.upload.calls)
+        self.assertEqual([((builds[1].build_request, ), {})],
+                         client.uploadManifestList.calls)
+
+    def test_failing_upload_does_not_retries_automatically(self):
+        build_request = self.makeBuildRequest(include_i386=False)
+        builds = build_request.builds
+        self.assertEqual(1, builds.count())
+
+        build = builds.one()
+        self.assertContentEqual([], build.registry_upload_jobs)
+        upload_job = OCIRegistryUploadJob.create(build)
+
+        client = mock.Mock()
+        client.upload.side_effect = Exception("Nope! Error.")
+        self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
+
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            JobRunner([upload_job]).runAll()
+        self.assertEqual(1, client.upload.call_count)
+        self.assertEqual(0, client.uploadManifestList.call_count)
+        self.assertEqual(JobStatus.FAILED, upload_job.status)
+        self.assertFalse(upload_job.build_uploaded)
+
+    def test_failing_upload_manifest_list_retries(self):
+        build_request = self.makeBuildRequest(include_i386=False)
+        builds = build_request.builds
+        self.assertEqual(1, builds.count())
+
+        build = builds.one()
+        self.assertContentEqual([], build.registry_upload_jobs)
+        upload_job = OCIRegistryUploadJob.create(build)
+
+        client = mock.Mock()
+        client.uploadManifestList.side_effect = (
+            OCIRegistryUploadJob.ManifestListUploadError("Nope! Error."))
+        self.useFixture(ZopeUtilityFixture(client, IOCIRegistryClient))
+
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            JobRunner([upload_job]).runAll()
+        self.assertEqual(1, client.upload.call_count)
+        self.assertEqual(1, client.uploadManifestList.call_count)
+        self.assertEqual(JobStatus.WAITING, upload_job.status)
+        self.assertTrue(upload_job.is_pending)
+        self.assertTrue(upload_job.build_uploaded)
+
+        # Retry should skip client.upload and only run
+        # client.uploadManifestList:
+        client.uploadManifestList.side_effect = None
+        with dbuser(config.IOCIRegistryUploadJobSource.dbuser):
+            JobRunner([upload_job]).runAll()
+        self.assertEqual(1, client.upload.call_count)
+        self.assertEqual(2, client.uploadManifestList.call_count)
+        self.assertEqual(JobStatus.COMPLETED, upload_job.status)
+        self.assertTrue(upload_job.build_uploaded)
+
+    def test_allBuildsUploaded_lock_between_two_jobs(self):
+        """Simple test to ensure that allBuildsUploaded method locks
+        rows in the database and make concurrent calls wait for that.
+
+        This is not a 100% reliable way to check that concurrent calls to
+        allBuildsUploaded will queue up since it relies on the
+        execution time, but it's a "good enough" approach: this test might
+        pass if the machine running it is *really, really* slow, but a failure
+        here will indicate that something is for sure wrong.
+        """
+
+        class AllBuildsUploadedChecker(threading.Thread):
+            """Thread to run upload_job.allBuildsUploaded tracking the time."""
+            def __init__(self, build_request):
+                super(AllBuildsUploadedChecker, self).__init__()
+                self.build_request = build_request
+                self.upload_job = None
+                # Locks the measurement start until we finished running the
+                # bootstrap code. Parent thread should call waitBootstrap
+                # after self.start().
+                self.bootstrap_lock = threading.Lock()
+                self.bootstrap_lock.acquire()
+                self.result = None
+                self.error = None
+                self.start_date = None
+                self.end_date = None
+
+            @property
+            def lock_duration(self):
+                return self.end_date - self.start_date
+
+            def waitBootstrap(self):
+                """Wait until self.bootstrap finishes running."""
+                self.bootstrap_lock.acquire()
+                # We don't actually need the lock... just wanted to wait
+                # for it. let's release it then.
+                self.bootstrap_lock.release()
+
+            def bootstrap(self):
+                try:
+                    build = self.build_request.builds[1]
+                    self.upload_job = OCIRegistryUploadJob.create(build)
+                finally:
+                    self.bootstrap_lock.release()
+
+            def run(self):
+                with admin_logged_in():
+                    self.bootstrap()
+                    self.start_date = datetime.now()
+                    try:
+                        self.result = self.upload_job.allBuildsUploaded(
+                            self.build_request)
+                    except Exception as e:
+                        self.error = e
+                    self.end_date = datetime.now()
+
+        # Create a build request with 2 builds.
+        build_request = self.makeBuildRequest()
+        builds = build_request.builds
+        self.assertEqual(2, builds.count())
+
+        # Create the upload job for the first build.
+        upload_job1 = OCIRegistryUploadJob.create(builds[0])
+        upload_job1 = removeSecurityProxy(upload_job1)
+
+        # How long the lock will be held by the first job, in seconds.
+        # Adjust to minimize false positives: a number too small here might
+        # make the test pass even if the lock is not correctly implemented.
+        # A number too big will slow down the test execution...
+        waiting_time = 2
+        # Start a clean transaction and lock the rows at database level.
+        transaction.commit()
+        self.assertFalse(upload_job1.allBuildsUploaded(build_request))
+
+        # Start, in parallel, another upload job to run `allBuildsUploaded`.
+        concurrent_checker = AllBuildsUploadedChecker(build_request)
+        concurrent_checker.start()
+        # Wait until concurrent_checker is ready to measure the time waiting
+        # for the database lock.
+        concurrent_checker.waitBootstrap()
+
+        # Wait a bit and release the database lock by committing current
+        # transaction.
+        time.sleep(waiting_time)
+        # Let's force the first job to be finished, just to make sure the
+        # second job will realise it's the last one running.
+        upload_job1.start()
+        upload_job1.complete()
+        transaction.commit()
+
+        # Now, the concurrent checker should have already finished running,
+        # without any error and it should have taken at least the
+        # waiting_time to finish running (since it was waiting).
+        concurrent_checker.join()
+        self.assertIsNone(concurrent_checker.error)
+        self.assertTrue(concurrent_checker.result)
+        self.assertGreaterEqual(
+            concurrent_checker.lock_duration, timedelta(seconds=waiting_time))
+
     def test_run_failed_registry_error(self):
         # A run that fails with a registry error sets the registry upload
         # status to FAILED, and stores the detailed errors.
diff --git a/lib/lp/oci/tests/test_ociregistryclient.py b/lib/lp/oci/tests/test_ociregistryclient.py
index 3c5aa8c..8de88b4 100644
--- a/lib/lp/oci/tests/test_ociregistryclient.py
+++ b/lib/lp/oci/tests/test_ociregistryclient.py
@@ -11,6 +11,7 @@ from functools import partial
 import io
 import json
 import os
+import re
 import tarfile
 import uuid
 
@@ -33,13 +34,17 @@ from testtools.matchers import (
     Raises,
     )
 import transaction
+from zope.component import getUtility
 from zope.security.proxy import removeSecurityProxy
 
+from lp.buildmaster.interfaces.processor import IProcessorSet
+from lp.oci.interfaces.ocirecipejob import IOCIRecipeRequestBuildsJobSource
 from lp.oci.interfaces.ociregistryclient import (
     BlobUploadFailed,
     ManifestUploadFailed,
     MultipleOCIRegistryError,
     )
+from lp.oci.model.ocirecipe import OCIRecipeBuildRequest
 from lp.oci.model.ociregistryclient import (
     BearerTokenRegistryClient,
     OCIRegistryAuthenticationError,
@@ -50,6 +55,7 @@ from lp.oci.model.ociregistryclient import (
 from lp.oci.tests.helpers import OCIConfigHelperMixin
 from lp.services.compat import mock
 from lp.testing import TestCaseWithFactory
+from lp.testing.fixture import ZopeUtilityFixture
 from lp.testing.layers import (
     DatabaseFunctionalLayer,
     LaunchpadZopelessLayer,
@@ -142,6 +148,23 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
 
         transaction.commit()
 
+    def addManifestResponses(self, push_rule, status_code=201, json=None):
+        """Add responses for manifest upload URLs."""
+        # PUT to "anonymous" architecture-specific manifest.
+        manifests_url = "{}/v2/{}/manifests/sha256:.*".format(
+            push_rule.registry_credentials.url,
+            push_rule.image_name
+        )
+        responses.add(
+            "PUT", re.compile(manifests_url), status=status_code, json=json)
+
+        # PUT to tagged multi-arch manifest.
+        manifests_url = "{}/v2/{}/manifests/edge".format(
+            push_rule.registry_credentials.url,
+            push_rule.image_name
+        )
+        responses.add("PUT", manifests_url, status=status_code, json=json)
+
     @responses.activate
     def test_upload(self):
         self._makeFiles()
@@ -154,11 +177,7 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         push_rule = self.build.recipe.push_rules[0]
         responses.add("GET", "%s/v2/" % push_rule.registry_url, status=200)
 
-        manifests_url = "{}/v2/{}/manifests/edge".format(
-            push_rule.registry_credentials.url,
-            push_rule.image_name
-        )
-        responses.add("PUT", manifests_url, status=201)
+        self.addManifestResponses(push_rule)
 
         self.client.upload(self.build)
 
@@ -196,7 +215,8 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         _upload_fixture = self.useFixture(MockPatch(
             "lp.oci.model.ociregistryclient.OCIRegistryClient._upload"))
         self.useFixture(MockPatch(
-            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer"))
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer",
+            return_value=999))
 
         self.push_rule.registry_credentials.setCredentials({
             "username": "test-username",
@@ -206,9 +226,7 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         push_rule = self.build.recipe.push_rules[0]
         responses.add("GET", "%s/v2/" % push_rule.registry_url, status=200)
 
-        manifests_url = "{}/v2/{}/manifests/edge".format(
-            push_rule.registry_credentials.url, push_rule.image_name)
-        responses.add("PUT", manifests_url, status=201)
+        self.addManifestResponses(push_rule)
 
         self.client.upload(self.build)
 
@@ -222,7 +240,8 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         upload_fixture = self.useFixture(MockPatch(
             "lp.oci.model.ociregistryclient.OCIRegistryClient._upload"))
         self.useFixture(MockPatch(
-            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer"))
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer",
+            return_value=999))
 
         push_rules = [
             self.push_rule,
@@ -237,10 +256,8 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
             responses.add(
                 "GET", "%s/v2/" % push_rule.registry_url, status=200)
 
-            manifests_url = "{}/v2/{}/manifests/edge".format(
-                push_rule.registry_credentials.url, push_rule.image_name)
             status = 400 if i < 2 else 201
-            responses.add("PUT", manifests_url, status=status)
+            self.addManifestResponses(push_rule, status_code=status)
 
         error = self.assertRaises(
             MultipleOCIRegistryError, self.client.upload, self.build)
@@ -466,15 +483,13 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         self.useFixture(MockPatch(
             "lp.oci.model.ociregistryclient.OCIRegistryClient._upload"))
         self.useFixture(MockPatch(
-            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer"))
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer",
+            return_value=999))
 
         push_rule = self.build.recipe.push_rules[0]
         responses.add(
             "GET", "{}/v2/".format(push_rule.registry_url), status=200)
 
-        manifests_url = "{}/v2/{}/manifests/edge".format(
-            push_rule.registry_credentials.url,
-            push_rule.image_name)
         put_errors = [
             {
                 "code": "MANIFEST_INVALID",
@@ -482,8 +497,8 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
                 "detail": [],
                 },
             ]
-        responses.add(
-            "PUT", manifests_url, status=400, json={"errors": put_errors})
+        self.addManifestResponses(
+            push_rule, status_code=400, json={"errors": put_errors})
 
         expected_msg = "Failed to upload manifest for {} ({}) in {}".format(
             self.build.recipe.name, self.push_rule.registry_url, self.build.id)
@@ -503,16 +518,14 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
         self.useFixture(MockPatch(
             "lp.oci.model.ociregistryclient.OCIRegistryClient._upload"))
         self.useFixture(MockPatch(
-            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer"))
+            "lp.oci.model.ociregistryclient.OCIRegistryClient._upload_layer",
+            return_value=999))
 
         push_rule = self.build.recipe.push_rules[0]
         responses.add(
             "GET", "{}/v2/".format(push_rule.registry_url), status=200)
 
-        manifests_url = "{}/v2/{}/manifests/edge".format(
-            push_rule.registry_credentials.url,
-            push_rule.image_name)
-        responses.add("PUT", manifests_url, status=200)
+        self.addManifestResponses(push_rule, status_code=200)
 
         expected_msg = "Failed to upload manifest for {} ({}) in {}".format(
             self.build.recipe.name, self.push_rule.registry_url, self.build.id)
@@ -526,6 +539,64 @@ class TestOCIRegistryClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
                         Equals(expected_msg)),
                     MatchesStructure(errors=Is(None))))))
 
+    @responses.activate
+    def test_multi_arch_manifest_upload(self):
+        """Ensure that multi-arch manifest upload works and tags correctly
+        the uploaded image."""
+        # Creates a build request with 2 builds.
+        recipe = self.build.recipe
+        build1 = self.build
+        build2 = self.factory.makeOCIRecipeBuild(
+            recipe=recipe)
+        naked_build1 = removeSecurityProxy(build1)
+        naked_build2 = removeSecurityProxy(build1)
+        naked_build1.processor = getUtility(IProcessorSet).getByName('386')
+        naked_build2.processor = getUtility(IProcessorSet).getByName('amd64')
+
+        # Creates a mock IOCIRecipeRequestBuildsJobSource, as it was created
+        # by the celery job and triggered the 2 registry uploads already.
+        job = mock.Mock()
+        job.builds = [build1, build2]
+        job.uploaded_manifests = {
+            build1.id: {"digest": "build1digest", "size": 123},
+            build2.id: {"digest": "build2digest", "size": 321},
+        }
+        job_source = mock.Mock()
+        job_source.getByOCIRecipeAndID.return_value = job
+        self.useFixture(
+            ZopeUtilityFixture(job_source, IOCIRecipeRequestBuildsJobSource))
+        build_request = OCIRecipeBuildRequest(recipe, -1)
+
+        push_rule = self.build.recipe.push_rules[0]
+        responses.add(
+            "GET", "{}/v2/".format(push_rule.registry_url), status=200)
+        self.addManifestResponses(push_rule, status_code=201)
+
+        self.client.uploadManifestList(build_request)
+        self.assertEqual(2, len(responses.calls))
+        auth_call, manifest_call = responses.calls
+        self.assertEndsWith(
+            manifest_call.request.url,
+            "/v2/%s/manifests/edge" % push_rule.image_name)
+        self.assertEqual({
+            "schemaVersion": 2,
+            "mediaType": "application/"
+                         "vnd.docker.distribution.manifest.list.v2+json",
+            "manifests": [{
+                "platform": {"os": "linux", "architecture": "amd64"},
+                "mediaType": "application/"
+                             "vnd.docker.distribution.manifest.v2+json",
+                "digest": "build1digest",
+                "size": 123
+            }, {
+                "platform": {"os": "linux", "architecture": "386"},
+                "mediaType": "application/"
+                             "vnd.docker.distribution.manifest.v2+json",
+                "digest": "build2digest",
+                "size": 321
+            }]
+        }, json.loads(manifest_call.request.body))
+
 
 class TestRegistryHTTPClient(OCIConfigHelperMixin, SpyProxyCallsMixin,
                              TestCaseWithFactory):