apps/cloud-image-uploader
0
0
Fork 0
Code Issues 11 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Support publishing container images to registries (#4)

Browse source 
This adds support for publishing container images to registries.
It is intended to replace the sync-latest-container-base-image.sh
and sync-ostree-base-containers.sh scripts we currently use for
this purpose.

It finds "docker" or "ociarchive" type images in the compose and
matches them against a list of known "repos" (fedora-toolbox,
fedora-silverblue etc.) Images that match are pushed to the
registry using `skopeo copy`. Then a manifest is produced from
the published images, and published itself; if the compose is a
Rawhide compose, or a compose of the current stable release, a
manifest is also published under an alias ("rawhide" or "latest",
respectively).

This initial implementation is intentionally closely based on the
sync-latest-container-base-image.sh approach. The other script
uses a somewhat different approach where the release numbered
manifest is created from the local image files and then pushed
with `--all`, which also causes the image files to be pushed,
then the manifest is copied to the 'aliased' name. I think the
eventual outcome is really the same in both cases, though.

The container support can be activated by configuring at least
one registry in the consumer config. If this is not done,
container images will not be handled.

As part of this, we move to trying every handler against every
image in the compose, and filter out non-cloud and non-container
images in the handlers themselves. This avoids having two levels
of filtering and makes the code a bit cleaner. We also drop the
`CantHandle` and `NotHandled` exceptions and concepts, as we've
decided the model is "any number of handlers might potentially
all handle a given image", so it's not really possible for an
individual handler to declare that an image is "not handled" by
this project at all. This means there's no need to distinguish
between different cases when a handler bails early, so it can
just `return` and we can drop the exception handling.

We also set an environment variable that disables caching
of filelists and PDC queries in fedfind (5.3.0+), as this can
cause things to be left out of pyvcr cassettes.

Signed-off-by: Adam Williamson <awilliam@redhat.com>
This commit is contained in:
Adam Williamson 2024-05-22 16:04:19 -07:00
commit bd6b7ab0b2
8 changed files with 3131 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

4
Containerfile
View file

@ -5,7 +5,9 @@ LABEL org.opencontainers.image.authors="Fedora Cloud SIG <cloud@lists.fedoraproj
RUN dnf install -y \
patch \
python3-pip \
python3-hatchling
python3-hatchling \
skopeo \
buildah
RUN mkdir -p /srv/cloud-uploader/
COPY . /srv/cloud-uploader/src

4
README.md
View file

@ -1,9 +1,11 @@
# cloud-image-uploader
An AMQP consumer that automatically uploads Cloud images to their respective homes.
An AMQP consumer that automatically uploads Cloud and container images to their respective homes.
## Configuring
The `fedora-messaging` service is the container entrypoint and is also used to provide most of the configuration. An example configuration file is included and the full list of options are in [fedora-messaging's configuration documentation](https://fedora-messaging.readthedocs.io/en/stable/user-guide/configuration.html). The `FEDORA_MESSAGING_CONF` environment variable should be set to the configuration file's location.
In addition to `fedora-messaging`, the [Azure Ansible collection](https://docs.ansible.com/ansible/latest/collections/azure/azcollection/) needs Azure credentials provided. It supports using environment variables, a credentials file at `~/.azure/credentials` or the `azure-cli` credentials. For example, to authenticate via environment variables using an app registration in the Microsoft Entra ID service with a client secret, setting the `AZURE_TENANT`, `AZURE_CLIENT_ID`, and `AZURE_SECRET` along with the `AZURE_SUBSCRIPTION_ID` will be picked up by the Ansible playbook. The app registration needs to be configured in the given subscription's access control settings.
Container image upload to registries uses the `skopeo` and `buildah` commands, so for this to work, the system must be in a state such that these commands have the necessary credentials to upload to the registries configured, non-interactively. If no container registries are configured, container images will not be handled.

3
fedora-messaging.toml.example
View file

@ -53,6 +53,9 @@ name = "centralus"
regional_replica_count = 2
storage_account_type = "Standard_LRS"
[consumer_config.container]
registries = ["registry.fedoraproject.org", "quay.io/fedora"]
[qos]
prefetch_size = 0
prefetch_count = 25

143
fedora_cloud_image_uploader/handler.py
View file

@ -3,8 +3,10 @@ import hashlib
import logging
import lzma
import os
import subprocess
import tempfile
import time
from collections.abc import Iterable
from typing import NamedTuple
import ansible_runner
@ -25,15 +27,22 @@ from . import PLAYBOOKS
_log = logging.getLogger(__name__)
class CantHandle(Exception):
"""An exception meaning this handler does not handle this image."""
class NotHandled(Exception):
"""
An exception meaning this is the right handler, but we do not handle this
particular image.
"""
def _run(args: Iterable[str]):
"""Run a command and handle errors."""
_log.debug("image_uploader running command %s", " ".join(args))
try:
ret = subprocess.run(args, encoding="utf-8", capture_output=True, timeout=7200)
except subprocess.TimeoutExpired:
_log.error("Command: %s timed out after two hours", " ".join(args))
raise fm_exceptions.Nack()
except OSError as err:
_log.error("Command: %s caused error %s", " ".join(args), err)
raise fm_exceptions.Nack()
if ret.returncode:
_log.error("Command: %s returned %d", " ".join(args), ret.returncode)
_log.error("stdout: %s", ret.stdout)
_log.error("stderr: %s", ret.stderr)
raise fm_exceptions.Nack()
class ReleaseInfo(NamedTuple):
@ -54,7 +63,10 @@ class Uploader:
self.requests = Session()
retry_config = Retry(total=5, backoff_factor=1)
self.requests.mount("https://", adapters.HTTPAdapter(max_retries=retry_config))
self.cloud_handlers = (self.handle_azure,)
self.handlers = (self.handle_azure, self.handle_container)
# tracks the container repos we got images for, for manifest
# creation purposes
self.container_repos = dict()
def __call__(self, message: fm_message.Message):
"""
@ -76,22 +88,13 @@ class Uploader:
except ff_exceptions.UnsupportedComposeError:
_log.info("Skipping compose %s as it contains no images", compose_id)
return
# reset for each message
self.container_repos = dict()
relinfo = self.get_relinfo(compose)
cloud_images = [img for img in compose.all_images if img["subvariant"] == "Cloud_Base"]
try:
for image in cloud_images:
for handler in self.cloud_handlers:
try:
handler(image, relinfo)
break
except CantHandle:
# try the next handler
pass
except NotHandled:
# this means we do not handle the image, go to next image
break
_log.debug("Missing handler for '%s'", image["type"])
for image in compose.all_images:
for handler in self.handlers:
handler(image, relinfo)
except fm_exceptions.Nack:
# If we failed to process an image, it's not likely the failure will resolve
# itself in the time it takes to re-queue the message and then consume it again.
@ -99,6 +102,41 @@ class Uploader:
time.sleep(60)
raise
if self.container_repos:
# manifest stuff
for repo in self.container_repos:
for registry in self.conf["container"]["registries"]:
# something like "registry.fedoraproject.org/fedora:40"
regname = f"{registry}/{repo}:{str(relinfo.relnum)}"
targets = [regname]
# we also create aliased manifests for rawhide and
# latest stable
if compose.release.lower() == "rawhide":
targets.append(f"{registry}/{repo}:rawhide")
elif relinfo.relnum == ff_helpers.get_current_release(branched=False):
targets.append(f"{registry}/{repo}:latest")
for target in targets:
# wipe the manifest if it exists already
_run(("buildah", "rmi", target))
# create the manifest with all arches
createargs = ["buildah", "manifest", "create", target]
createargs.extend(
# it's intentional that this is regname not target
f"{regname}-{arch}"
for arch in self.container_repos[repo]
)
_run(createargs)
# push it
pushargs = (
"buildah",
"manifest",
"push",
target,
f"docker://{target}",
"--all",
)
_run(pushargs)
def get_relinfo(self, ffrel: ff_release.Release):
"""
Return a namedtuple that acts as a container for some useful
@ -212,12 +250,14 @@ class Uploader:
"""
Handle Azure images.
"""
if image["type"] != "vhd-compressed":
raise CantHandle()
if image.get("subvariant") != "Cloud_Base" or image.get("type") != "vhd-compressed":
return
if image["arch"] not in ("x86_64", "aarch64"):
raise NotHandled("Unsupported arch")
# unsupported arch
return
if relinfo.relnum < 40:
raise NotHandled("Images prior to F40 aren't supported")
# images prior to F40 aren't supported
return
with tempfile.TemporaryDirectory() as workdir:
image_path = self.download_image(image, workdir, decompress=True)
@ -343,3 +383,50 @@ class Uploader:
_log.info(
f"Deleted image definition {image_definition.name} since it has no versions"
)
def handle_container(self, image: dict, relinfo: ReleaseInfo):
"""Handle container images."""
registries = self.conf.get("container", {}).get("registries")
if not registries:
# we can't do anything if no registries are configured
return
if image["type"] not in ("docker", "ociarchive"):
# not a known container image type
return
repos = {
"Container_Toolbox": "fedora-toolbox",
"Container_Minimal_Base": "fedora-minimal",
"Container_Base": "fedora",
"Silverblue": "fedora-silverblue",
"Kinoite": "fedora-kinoite",
"Onyx": "fedora-onyx",
"Sericea": "fedora-sericea",
}
repo = repos.get(image["subvariant"])
if not repo:
_log.debug("Unknown subvariant %s", image["subvariant"])
return
if image["type"] == "docker" and relinfo.relnum < 40:
# these are actual docker archive images
imgformat = "docker-archive"
else:
# all others are OCI archives; F40+ .oci.tar.xz images
# with type "docker" are xz-compressed OCI archives,
# .ociarchive images with type "ociarchive" are non-
# compressed OCI archives
imgformat = "oci-archive"
arch = image["arch"]
with tempfile.TemporaryDirectory() as workdir:
image_path = self.download_image(image, workdir, decompress=True)
for registry in registries:
args = [
"skopeo",
"copy",
f"{imgformat}:{image_path}",
f"docker://{registry}/{repo}:{str(relinfo.relnum)}-{arch}",
]
_run(args)
if repo in self.container_repos:
self.container_repos[repo].append(arch)
else:
self.container_repos[repo] = [arch]

683
tests/fixtures/cassettes/test_containers[compose0].yaml vendored Normal file
View file

File diff suppressed because one or more lines are too long

1450
tests/fixtures/cassettes/test_containers[compose1].yaml vendored Normal file
View file

File diff suppressed because it is too large Load diff

683
tests/fixtures/cassettes/test_containers_registries_not_configured.yaml vendored Normal file
View file

File diff suppressed because one or more lines are too long

223
tests/test_handler.py
View file

@ -19,11 +19,30 @@ from freezegun import freeze_time
from requests.exceptions import RequestException
from fedora_cloud_image_uploader import PLAYBOOKS, Uploader
from fedora_cloud_image_uploader.handler import CantHandle, NotHandled
from fedora_cloud_image_uploader.handler import _run
# disable fedfind caching, as it can cause things to be left out of
# pyvcr cassettes
os.environ["FEDFIND_NO_CACHE"] = "1"
def _mock_download_image(self, image: dict, dest_dir: str, decompress=False) -> str:
"""
A mocked-out download_image that behaves somewhat like the real
one.
"""
fn = os.path.basename(image["path"])
if decompress and fn.endswith(".xz"):
fn = fn.removesuffix(".xz")
return os.path.join(dest_dir, fn)
@mock.patch("fedora_cloud_image_uploader.handler.ansible_runner")
@pytest.mark.vcr
@mock.patch(
"fedora_cloud_image_uploader.handler.Uploader.download_image",
lambda a, b, c, decompress: f"/test/{os.path.basename(b['path'].removesuffix('.xz'))}",
)
@mock.patch("fedora_cloud_image_uploader.handler.ansible_runner")
@pytest.mark.parametrize(
"compose",
[
@ -51,7 +70,8 @@ def test_gallery_name(mock_runner, fixtures_dir, compose):
}
consumer = Uploader()
consumer.download_image = mock.Mock()
# disable handlers we don't want to hit in this test
consumer.handlers = [consumer.handle_azure]
consumer(msg)
assert mock_runner.interface.run.call_count == 2
@ -63,6 +83,133 @@ def test_gallery_name(mock_runner, fixtures_dir, compose):
@pytest.mark.vcr
@mock.patch(
"fedora_cloud_image_uploader.handler.Uploader.download_image",
lambda a, b, c, decompress: f"/test/{os.path.basename(b['path'].removesuffix('.xz'))}",
)
@mock.patch("subprocess.run")
@pytest.mark.parametrize(
"compose",
[
(
"messages/rawhide_compose.json",
"Rawhide-20240501.n.0",
"41",
"rawhide",
{
"fedora-minimal": ["aarch64", "x86_64", "s390x", "ppc64le"],
"fedora": ["aarch64", "x86_64", "s390x", "ppc64le"],
"fedora-toolbox": ["aarch64", "x86_64", "s390x", "ppc64le"],
},
),
(
"messages/rc_compose.json",
"40-1.14",
"40",
"latest",
{
"fedora-minimal": ["aarch64", "ppc64le", "s390x", "x86_64"],
"fedora": ["aarch64", "ppc64le", "s390x", "x86_64"],
"fedora-toolbox": ["aarch64", "ppc64le", "s390x", "x86_64"],
"fedora-kinoite": ["aarch64", "ppc64le", "x86_64"],
"fedora-onyx": ["x86_64"],
"fedora-sericea": ["aarch64", "x86_64"],
"fedora-silverblue": ["aarch64", "ppc64le", "x86_64"],
},
),
],
)
def test_containers(mock_subrun, fixtures_dir, compose):
mock_subrun.return_value.returncode = 0
message_file, cidorlabel, relnum, alias, expected_images = compose
registries = ["registry.fedoraproject.org", "quay.io/fedora"]
# mapping of reponames to expected image filename base strings
repotoid = {
"fedora": "Fedora-Container-Base-Generic",
"fedora-minimal": "Fedora-Container-Base-Generic-Minimal",
"fedora-toolbox": "Fedora-Container-Toolbox",
"fedora-kinoite": "Fedora-Kinoite",
"fedora-onyx": "Fedora-Onyx",
"fedora-sericea": "Fedora-Sericea",
"fedora-silverblue": "Fedora-Silverblue",
}
with open(os.path.join(fixtures_dir, message_file)) as fd:
msg = message.load_message(json.load(fd))
config.conf["consumer_config"]["container"] = {"registries": registries}
consumer = Uploader()
# disable handlers we don't want to hit in this test
consumer.handlers = [consumer.handle_container]
consumer(msg)
# this gives us a list of strings representing the commands run
# (space-joined args iterables passed to _run)
# we will check that every command we expect is in here, remove
# them all, and assert it's empty at the end
calls = [" ".join(call.args[0]) for call in mock_subrun.call_args_list]
for registry in registries:
for exprepo, arches in expected_images.items():
# find the expected calls to skopeo copy
for arch in arches:
ident = repotoid[exprepo]
if "Container" in ident:
expfn = f"oci-archive:/test/{ident}.{arch}-{cidorlabel}.oci.tar"
else:
# atomic desktop filenames don't have the arch and
# use a non-standard compose label representation
# https://gitlab.com/fedora/ostree/sig/-/issues/31
expfn = f"oci-archive:/test/{ident}-{cidorlabel.replace('-', '.')}.ociarchive"
expcall = f"skopeo copy {expfn} docker://{registry}/{exprepo}:{relnum}-{arch}"
assert expcall in calls
calls.remove(expcall)
repopath = f"{registry}/{exprepo}:{relnum}"
aliaspath = f"{registry}/{exprepo}:{alias}"
# find the expected calls to buildah
expcalls = (
f"buildah rmi {repopath}",
f"buildah rmi {aliaspath}",
f"buildah manifest create {repopath} "
+ " ".join(f"{repopath}-{arch}" for arch in arches),
f"buildah manifest create {aliaspath} "
+ " ".join(f"{repopath}-{arch}" for arch in arches),
f"buildah manifest push {repopath} docker://{repopath} --all",
f"buildah manifest push {aliaspath} docker://{aliaspath} --all",
)
for call in expcalls:
assert call in calls
calls.remove(call)
assert len(calls) == 0
@pytest.mark.vcr
@mock.patch("subprocess.run")
@mock.patch("fedora_cloud_image_uploader.handler.Uploader.download_image")
def test_containers_registries_not_configured(mock_dl, mock_run, fixtures_dir):
"""
Test we correctly skip container handling if registries are not
configured.
"""
with open(os.path.join(fixtures_dir, "messages/rawhide_compose.json")) as fd:
msg = message.load_message(json.load(fd))
config.conf["consumer_config"]["container"] = {}
consumer = Uploader()
# disable handlers we don't want to hit in this test
consumer.handlers = [consumer.handle_container]
consumer(msg)
assert mock_dl.call_count == 0
assert mock_run.call_count == 0
@pytest.mark.vcr
@mock.patch(
"fedora_cloud_image_uploader.handler.Uploader.download_image",
lambda a, b, c, decompress: f"/test/{os.path.basename(b['path'].removesuffix('.xz'))}",
)
@mock.patch("fedora_cloud_image_uploader.handler.ansible_runner")
def test_old_unsupported_azure_compose(mock_runner, fixtures_dir):
mock_runner.interface.run.return_value.rc = 0
@ -78,9 +225,11 @@ def test_old_unsupported_azure_compose(mock_runner, fixtures_dir):
"storage_container_name": "vhds",
"target_regions": {},
}
config.conf["consumer_config"]["container"] = {
"registries": ["registry.fedoraproject.org", "quay.io/fedora"]
}
consumer = Uploader()
consumer.download_image = mock.Mock()
consumer(msg)
assert mock_runner.interface.run.call_count == 0
@ -128,25 +277,26 @@ def test_ansible_fail(mock_run, caplog):
assert caplog.records[-1].msg == "Playbook failed with return code 1"
def test_azure_filters():
@mock.patch("fedora_cloud_image_uploader.handler.ansible_runner")
def test_azure_filters(mock_runner):
"""Test the cases where AzureHandler should decide not to handle."""
config.conf["consumer_config"]["azure"] = {}
relinfo = mock.MagicMock()
relinfo.relnum = 40
image = {"type": "notonewelike", "arch": "x86_64"}
image = {"type": "notonewelike", "arch": "x86_64", "subvariant": "Cloud_Base"}
consumer = Uploader()
with pytest.raises(CantHandle):
consumer.handle_azure(image, relinfo)
consumer.handle_azure(image, relinfo)
assert mock_runner.call_count == 0
image["type"] = "vhd-compressed"
image["arch"] = "ppc64le"
with pytest.raises(NotHandled) as excinfo:
consumer.handle_azure(image, relinfo)
assert str(excinfo.value) == "Unsupported arch"
consumer.handle_azure(image, relinfo)
assert mock_runner.call_count == 0
image["arch"] = "x86_64"
relinfo.relnum = 39
with pytest.raises(NotHandled) as excinfo:
consumer.handle_azure(image, relinfo)
assert str(excinfo.value) == "Images prior to F40 aren't supported"
consumer.handle_azure(image, relinfo)
assert mock_runner.call_count == 0
@mock.patch("fedora_cloud_image_uploader.handler.ansible_runner")
@ -189,25 +339,13 @@ def test_consumer_handle_exceptions(mock_getrel, mock_relinfo, mock_sleep, fixtu
consumer = Uploader()
ffrel = mock_getrel.return_value
ffrel.all_images = [{"subvariant": "Cloud_Base", "type": "foo"}]
ffrel.all_images = [{"subvariant": "Cloud_Base", "type": "foo", "format": "foo"}]
mock_handler1 = mock.MagicMock()
mock_handler1.side_effect = CantHandle
mock_handler2 = mock.MagicMock()
with mock.patch.object(consumer, "cloud_handlers", [mock_handler1, mock_handler2]):
consumer(msg)
# we should have continued to the second handler for CantHandle
assert mock_handler2.call_count == 1
mock_handler2.reset_mock()
mock_handler1.side_effect = NotHandled
with mock.patch.object(consumer, "cloud_handlers", [mock_handler1, mock_handler2]):
consumer(msg)
# for NotHandled we should bail out and never reach second handler
assert mock_handler2.call_count == 0
mock_sleep.reset_mock()
mock_handler1.side_effect = exceptions.Nack
with mock.patch.object(consumer, "cloud_handlers", [mock_handler1, mock_handler2]):
with pytest.raises(exceptions.Nack):
consumer(msg)
mock_handler2 = mock.MagicMock()
consumer.handlers = [mock_handler1, mock_handler2]
with pytest.raises(exceptions.Nack):
consumer(msg)
assert mock_handler2.call_count == 0
# we should sleep before re-raising the Nack
assert mock_sleep.call_count == 1
@ -389,3 +527,26 @@ def test_azure_rawhide_images(mock_az_client, azure_env_vars, azure_fm_conf):
expected_calls.reverse()
actual_calls = [call for call in client.gallery_image_versions.begin_delete.call_args_list]
assert expected_calls == actual_calls
def test_run_error(caplog):
"""Test the error handling in _run."""
with pytest.raises(exceptions.Nack):
_run(("ls", "/some/where/that/doesnt/exist"))
assert (
caplog.records[-1].getMessage()
== "stderr: ls: cannot access '/some/where/that/doesnt/exist': No such file or directory\n"
)
assert caplog.records[-2].getMessage() == "stdout: "
assert caplog.records[-3].getMessage() == "Command: ls /some/where/that/doesnt/exist returned 2"
with pytest.raises(exceptions.Nack):
_run(("commandthatdoesntexist",))
assert caplog.records[-1].getMessage() == (
"Command: commandthatdoesntexist caused error [Errno 2] "
"No such file or directory: 'commandthatdoesntexist'"
)
with pytest.raises(exceptions.Nack):
# this is a cute freeze_time trick to force a timeout
with freeze_time("2024-05-28", auto_tick_seconds=10000):
_run(("ls",))
assert caplog.records[-1].getMessage() == "Command: ls timed out after two hours"