23616668936

Committed 26 Mar 2026 08:33PM UTC coverage: 92.848% (-0.08%) from 92.923%

Build # 23616668936

Build Type

Pull #23133

github

Committed by

web-flow

Commit Message

Merge 51b4d6d01 into 9b3c1562e

Pull Request Pull Request #23133: Add buildctl engine

Coverage Stats

325 of 386 new or added lines in 14 files covered. (84.2%)

30 existing lines in 4 files now uncovered.

91645 of 98704 relevant lines covered (92.85%)

4.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.72

/src/python/pants/backend/docker/goals/package_image.py

# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import json
import logging
import os
import re
from collections.abc import Iterator
from dataclasses import asdict, dataclass
from functools import partial
from itertools import chain
from typing import Literal, cast

from pants.backend.docker.engine_types import DockerBuildEngine
from pants.backend.docker.package_types import (
    BuiltDockerImage,
    DockerPushOnPackageBehavior,
    DockerPushOnPackageException,
)
from pants.backend.docker.registries import DockerRegistries, DockerRegistryOptions
from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.target_types import (
    DockerImageBuildImageOutputField,
    DockerImageContextRootField,
    DockerImageRegistriesField,
    DockerImageRepositoryField,
    DockerImageSourceField,
    DockerImageTagsField,
    DockerImageTagsRequest,
    DockerImageTargetStageField,
    OptionValueFormatter,
    ValidateOptionsMixin,
    get_docker_image_tags,
)
from pants.backend.docker.util_rules.binaries import (
    BuildctlBinary,
    DockerBinary,
    PodmanBinary,
    get_buildctl,
    get_docker,
    get_podman,
)
from pants.backend.docker.util_rules.docker_build_context import (
    DockerBuildContext,
    DockerBuildContextRequest,
    create_docker_build_context,
)
from pants.backend.docker.utils import format_rename_suggestion
from pants.core.goals.package import BuiltPackage, OutputPathField, PackageFieldSet
from pants.core.goals.publish import PublishFieldSet
from pants.engine.collection import Collection
from pants.engine.fs import EMPTY_DIGEST, CreateDigest, FileContent
from pants.engine.internals.graph import resolve_target
from pants.engine.intrinsics import create_digest, execute_process
from pants.engine.process import Process, ProcessExecutionFailure
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import InvalidFieldException, Target, WrappedTargetRequest
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.global_options import GlobalOptions, KeepSandboxes
from pants.util.strutil import bullet_list, softwrap
from pants.util.value_interpolation import InterpolationContext, InterpolationError

logger = logging.getLogger(__name__)


class DockerImageTagValueError(InterpolationError):
    pass


class DockerRepositoryNameError(InterpolationError):
    pass


class DockerBuildTargetStageError(ValueError):
    pass


class DockerImageOptionValueError(InterpolationError):
    pass


@dataclass(frozen=True)
class DockerPackageFieldSet(PackageFieldSet):
    required_fields = (DockerImageSourceField,)

    context_root: DockerImageContextRootField
    registries: DockerImageRegistriesField
    repository: DockerImageRepositoryField
    source: DockerImageSourceField
    tags: DockerImageTagsField
    target_stage: DockerImageTargetStageField
    output_path: OutputPathField
    output: DockerImageBuildImageOutputField

    def pushes_on_package(self) -> bool:
        """Returns True if this docker_image target would push to a registry during packaging."""
        return bool(self.output.value) and (
            self.output.value.get("push") == "true" or self.output.value["type"] == "registry"
        )

    def format_tag(self, tag: str, interpolation_context: InterpolationContext) -> str:
        source = InterpolationContext.TextSource(
            address=self.address, target_alias="docker_image", field_alias=self.tags.alias
        )
        return interpolation_context.format(tag, source=source, error_cls=DockerImageTagValueError)

    def format_repository(
        self,
        default_repository: str,
        interpolation_context: InterpolationContext,
        registry: DockerRegistryOptions | None = None,
    ) -> str:
        repository_context = InterpolationContext.from_dict(
            {
                "name": self.address.target_name,
                "directory": os.path.basename(self.address.spec_path),
                "full_directory": self.address.spec_path,
                "parent_directory": os.path.basename(os.path.dirname(self.address.spec_path)),
                "default_repository": default_repository,
                "target_repository": self.repository.value or default_repository,
                **interpolation_context,
            }
        )
        if registry and registry.repository:
            repository_text = registry.repository
            source = InterpolationContext.TextSource(
                options_scope=f"[docker.registries.{registry.alias or registry.address}].repository"
            )
        elif self.repository.value:
            repository_text = self.repository.value
            source = InterpolationContext.TextSource(
                address=self.address, target_alias="docker_image", field_alias=self.repository.alias
            )
        else:
            repository_text = default_repository
            source = InterpolationContext.TextSource(options_scope="[docker].default_repository")
        return repository_context.format(
            repository_text, source=source, error_cls=DockerRepositoryNameError
        ).lower()

    def format_image_ref_tags(
        self,
        repository: str,
        tags: tuple[str, ...],
        interpolation_context: InterpolationContext,
        uses_local_alias,
    ) -> Iterator[ImageRefTag]:
        for tag in tags:
            formatted = self.format_tag(tag, interpolation_context)
            yield ImageRefTag(
                template=tag,
                formatted=formatted,
                full_name=":".join(s for s in [repository, formatted] if s),
                uses_local_alias=uses_local_alias,
            )

    def image_refs(
        self,
        default_repository: str,
        registries: DockerRegistries,
        interpolation_context: InterpolationContext,
        additional_tags: tuple[str, ...] = (),
    ) -> Iterator[ImageRefRegistry]:
        """The per-registry image refs: each returned element is a collection of the tags applied to
        the image in a single registry.

        In the Docker world, the term `tag` is used both for what we here prefer to call the image
        `ref`, as well as for the image version, or tag, that is at the end of the image name
        separated with a colon. By introducing the image `ref` we can retain the use of `tag` for
        the version part of the image name.

        This function returns all image refs to apply to the Docker image, grouped by
        registry. Within each registry, the `tags` attribute contains a metadata about each tag in
        the context of that registry, and the `full_name` attribute of each `ImageRefTag` provides
        the image ref, of the following form:

            [<registry>/]<repository-name>[:<tag>]

        Where the `<repository-name>` may contain any number of separating slashes `/`, depending on
        the `default_repository` from configuration or the `repository` field on the target
        `docker_image`.

        This method will always return at least one `ImageRefRegistry`, and there will be at least
        one tag.
        """
        image_tags = (self.tags.value or ()) + additional_tags
        registries_options = tuple(registries.get(*(self.registries.value or [])))
        if not registries_options:
            # The image name is also valid as image ref without registry.
            repository = self.format_repository(default_repository, interpolation_context)
            yield ImageRefRegistry(
                registry=None,
                repository=repository,
                tags=tuple(
                    self.format_image_ref_tags(
                        repository, image_tags, interpolation_context, uses_local_alias=False
                    )
                ),
            )
            return

        for registry in registries_options:
            repository = self.format_repository(default_repository, interpolation_context, registry)
            address_repository = "/".join([registry.address, repository])
            if registry.use_local_alias and registry.alias:
                alias_repository = "/".join([registry.alias, repository])
            else:
                alias_repository = None

            yield ImageRefRegistry(
                registry=registry,
                repository=repository,
                tags=(
                    *self.format_image_ref_tags(
                        address_repository,
                        image_tags + registry.extra_image_tags,
                        interpolation_context,
                        uses_local_alias=False,
                    ),
                    *(
                        self.format_image_ref_tags(
                            alias_repository,
                            image_tags + registry.extra_image_tags,
                            interpolation_context,
                            uses_local_alias=True,
                        )
                        if alias_repository
                        else []
                    ),
                ),
            )

    def get_context_root(self, default_context_root: str) -> str:
        """Examines `default_context_root` and `self.context_root.value` and translates that to a
        context root for the Docker build operation.

        That is, in the configuration/field value, the context root is relative to build root when
        in the form `path/..` (implies semantics as `//path/..` for target addresses) or the BUILD
        file when `./path/..`.

        The returned path is always relative to the build root.
        """
        if self.context_root.value is not None:
            context_root = self.context_root.value
        else:
            context_root = cast(
                str, self.context_root.compute_value(default_context_root, self.address)
            )
        if context_root.startswith("./"):
            context_root = os.path.join(self.address.spec_path, context_root)
        return os.path.normpath(context_root)


@dataclass(frozen=True)
class ImageRefRegistry:
    registry: DockerRegistryOptions | None
    repository: str
    tags: tuple[ImageRefTag, ...]


@dataclass(frozen=True)
class ImageRefTag:
    template: str
    formatted: str
    full_name: str
    uses_local_alias: bool


@dataclass(frozen=True)
class DockerInfoV1:
    """The format of the `$target_name.docker-info.json` file."""

    version: Literal[1]
    image_id: str
    # It'd be good to include the digest here (e.g. to allow 'docker run
    # registry/repository@digest'), but that is only known after pushing to a V2 registry

    registries: list[DockerInfoV1Registry]

    @staticmethod
    def serialize(image_refs: tuple[ImageRefRegistry, ...], image_id: str) -> bytes:
        # make sure these are in a consistent order (the exact order doesn't matter
        # so much), no matter how they were configured
        sorted_refs = sorted(image_refs, key=lambda r: r.registry.address if r.registry else "")

        info = DockerInfoV1(
            version=1,
            image_id=image_id,
            registries=[
                DockerInfoV1Registry(
                    alias=r.registry.alias if r.registry and r.registry.alias else None,
                    address=r.registry.address if r.registry else None,
                    repository=r.repository,
                    tags=[
                        DockerInfoV1ImageTag(
                            template=t.template,
                            tag=t.formatted,
                            uses_local_alias=t.uses_local_alias,
                            name=t.full_name,
                        )
                        # consistent order, as above
                        for t in sorted(r.tags, key=lambda t: t.full_name)
                    ],
                )
                for r in sorted_refs
            ],
        )

        return json.dumps(asdict(info)).encode()


@dataclass(frozen=True)
class DockerInfoV1Registry:
    # set if registry was specified as `@something`
    alias: str | None
    address: str | None
    repository: str
    tags: list[DockerInfoV1ImageTag]


@dataclass(frozen=True)
class DockerInfoV1ImageTag:
    template: str
    tag: str
    uses_local_alias: bool
    # for convenience, include the concatenated registry/repository:tag name (using this tag)
    name: str


def get_value_formatter(
    context: DockerBuildContext, target: Target, field_alias: str
) -> OptionValueFormatter:
    return partial(
        context.interpolation_context.format,
        source=InterpolationContext.TextSource(
            address=target.address, target_alias=target.alias, field_alias=field_alias
        ),
        error_cls=DockerImageOptionValueError,
    )


def get_build_options(
    context: DockerBuildContext,
    docker_options: DockerOptions,
    target: Target,
) -> Iterator[str]:
    gen_options_func_name = (
        "buildctl_options"
        if docker_options.build_engine == DockerBuildEngine.BUILDCTL
        else "docker_build_options"
    )
    for field_type in target.field_types:
        if (
            issubclass(field_type, ValidateOptionsMixin)
            and target[field_type].validate_options(docker_options, context)
            and (gen_options_func := getattr(target[field_type], gen_options_func_name, None))
        ):
            yield from gen_options_func(
                docker=docker_options,
                value_formatter=get_value_formatter(context, target, field_type.alias),
            )

    # Special handling for global options
    if docker_options.build_target_stage in context.stages:
        compute_options_func = (
            DockerImageTargetStageField.compute_buildctl_options
            if docker_options.build_engine == DockerBuildEngine.BUILDCTL
            else DockerImageTargetStageField.compute_docker_build_options
        )
        yield from compute_options_func(
            docker_options.build_target_stage,
            docker=docker_options,
            value_formatter=get_value_formatter(context, target, DockerImageTargetStageField.alias),
        )

    # This is the same for docker and buildkit
    if docker_options.build_no_cache:
        yield "--no-cache"


@dataclass(frozen=True)
class GetImageRefsRequest:
    field_set: DockerPackageFieldSet
    build_upstream_images: bool


class DockerImageRefs(Collection[ImageRefRegistry]):
    pass


@rule
async def get_image_refs(
    request: GetImageRefsRequest, options: DockerOptions, union_membership: UnionMembership
) -> DockerImageRefs:
    context, wrapped_target = await concurrently(
        create_docker_build_context(
            DockerBuildContextRequest(
                address=request.field_set.address,
                build_upstream_images=request.build_upstream_images,
            ),
            **implicitly(),
        ),
        resolve_target(
            WrappedTargetRequest(request.field_set.address, description_of_origin="<infallible>"),
            **implicitly(),
        ),
    )

    image_tags_requests = union_membership.get(DockerImageTagsRequest)
    additional_image_tags = await concurrently(
        get_docker_image_tags(
            **implicitly({image_tags_request_cls(wrapped_target.target): DockerImageTagsRequest})
        )
        for image_tags_request_cls in image_tags_requests
        if image_tags_request_cls.is_applicable(wrapped_target.target)
    )

    return DockerImageRefs(
        request.field_set.image_refs(
            default_repository=options.default_repository,
            registries=options.registries(),
            interpolation_context=context.interpolation_context,
            additional_tags=tuple(chain.from_iterable(additional_image_tags)),
        )
    )


@dataclass(frozen=True)
class DockerImageBuildProcess:
    process: Process
    context: DockerBuildContext
    context_root: str
    image_refs: DockerImageRefs
    tags: tuple[str, ...]


@rule
async def get_docker_image_build_process(
    field_set: DockerPackageFieldSet, options: DockerOptions
) -> DockerImageBuildProcess:
    context, wrapped_target, image_refs = await concurrently(
        create_docker_build_context(
            DockerBuildContextRequest(
                address=field_set.address,
                build_upstream_images=True,
            ),
            **implicitly(),
        ),
        resolve_target(
            WrappedTargetRequest(field_set.address, description_of_origin="<infallible>"),
            **implicitly(),
        ),
        get_image_refs(
            GetImageRefsRequest(
                field_set=field_set,
                build_upstream_images=True,
            ),
            **implicitly(),
        ),
    )
    tags = tuple(tag.full_name for registry in image_refs for tag in registry.tags)
    if not tags:
        raise InvalidFieldException(
            softwrap(
                f"""
                The `{DockerImageTagsField.alias}` field in target {field_set.address} must not be
                empty, unless there is a custom plugin providing additional tags using the
                `DockerImageTagsRequest` union type.
                """
            )
        )

    # Mix the upstream image ids into the env to ensure that Pants invalidates this
    # image-building process correctly when an upstream image changes, even though the
    # process itself does not consume this data.
    env = {
        **context.build_env.environment,
        "__UPSTREAM_IMAGE_IDS": ",".join(context.upstream_image_ids),
    }
    context_root = field_set.get_context_root(options.default_context_root)
    binary: BuildctlBinary | PodmanBinary | DockerBinary
    match options.build_engine:
        case DockerBuildEngine.BUILDCTL:
            binary = await get_buildctl(**implicitly())
        case DockerBuildEngine.PODMAN:
            binary = await get_podman(**implicitly())
        case _:
            binary = await get_docker(**implicitly())

    process = binary.build_image(
        build_args=context.build_args,
        digest=context.digest,
        dockerfile=context.dockerfile,
        context_root=context_root,
        env=env,
        tags=tags,
        output=field_set.output.value,
        extra_args=tuple(
            get_build_options(
                context=context,
                docker_options=options,
                target=wrapped_target.target,
            )
        ),
        is_publish=isinstance(field_set, PublishFieldSet),
    )
    return DockerImageBuildProcess(
        process=process,
        context=context,
        context_root=context_root,
        image_refs=image_refs,
        tags=tags,
    )


@rule
async def build_docker_image(
    field_set: DockerPackageFieldSet,
    options: DockerOptions,
    global_options: GlobalOptions,
    keep_sandboxes: KeepSandboxes,
) -> BuiltPackage:
    """Build a Docker image using `docker build`."""
    # Check if this build would push and handle according to push_on_package behavior
    if field_set.pushes_on_package():
        match options.push_on_package:
            case DockerPushOnPackageBehavior.IGNORE:
                return BuiltPackage(EMPTY_DIGEST, ())
            case DockerPushOnPackageBehavior.ERROR:
                raise DockerPushOnPackageException(field_set.address)
            case DockerPushOnPackageBehavior.WARN:
                logger.warning(
                    f"Docker image {field_set.address} will push to a registry during packaging"
                )

    build_process = await get_docker_image_build_process(field_set, **implicitly())
    result = await execute_process(build_process.process, **implicitly())

    if result.exit_code != 0:
        msg = f"{options.build_engine.value.capitalize()} build failed for `docker_image` {field_set.address}."
        if options.suggest_renames:
            maybe_help_msg = format_docker_build_context_help_message(
                context_root=build_process.context_root,
                context=build_process.context,
                colors=global_options.colors,
            )
            if maybe_help_msg:
                msg += " " + maybe_help_msg

        logger.warning(msg)

        raise ProcessExecutionFailure(
            result.exit_code,
            result.stdout,
            result.stderr,
            build_process.process.description,
            keep_sandboxes=keep_sandboxes,
        )

    parse_image_id = (
        parse_image_id_from_podman_build_output
        if options.build_engine == DockerBuildEngine.PODMAN
        else parse_image_id_from_buildkit_output
    )
    image_id = parse_image_id(result.stdout, result.stderr) or "<unknown>"
    docker_build_output_msg = "\n".join(
        (
            f"{options.build_engine.value.capitalize()} build output for {build_process.tags[0]}:",
            "stdout:",
            result.stdout.decode(),
            "stderr:",
            result.stderr.decode(),
        )
    )

    if options.build_verbose:
        logger.info(docker_build_output_msg)
    else:
        logger.debug(docker_build_output_msg)

    metadata_filename = field_set.output_path.value_or_default(file_ending="docker-info.json")
    metadata = DockerInfoV1.serialize(build_process.image_refs, image_id=image_id)
    digest = await create_digest(CreateDigest([FileContent(metadata_filename, metadata)]))

    return BuiltPackage(
        digest,
        (BuiltDockerImage.create(image_id, build_process.tags, metadata_filename),),
    )


def parse_image_id_from_buildkit_output(*outputs: bytes) -> str | None:
    """Outputs are typically the stdout/stderr pair from the `docker build` process."""
    # NB: We use the extracted image id for invalidation. The short_id may theoretically
    #  not be unique enough, although in a non adversarial situation, this is highly unlikely
    #  to be an issue in practice.
    image_id_regexp = re.compile(
        "|".join(
            (
                # BuildKit output.
                r"(writing image (?P<digest>sha256:\S+))",
                # Buildkit with --push=true output.
                r"(pushing manifest for (?P<pushed_manifest>\S+))",
                # BuildKit with containerd-snapshotter output.
                r"(exporting manifest list (?P<manifest_list>sha256:\S+))",
                # BuildKit with containerd-snapshotter output and no attestation.
                r"(exporting manifest (?P<manifest>sha256:\S+))",
                # Docker output.
                r"(Successfully built (?P<short_id>\S+))",
            ),
        )
    )
    for output in outputs:
        image_id_match = next(
            (
                match
                for match in (
                    re.search(image_id_regexp, line)
                    for line in reversed(output.decode().split("\n"))
                )
                if match
            ),
            None,
        )
        if image_id_match:
            image_id = (
                image_id_match.group("digest")
                or image_id_match.group("pushed_manifest")
                or image_id_match.group("short_id")
                or image_id_match.group("manifest_list")
                or image_id_match.group("manifest")
            )
            return image_id

    return None


def parse_image_id_from_podman_build_output(*outputs: bytes) -> str | None:
    for output in outputs:
        try:
            _, image_id, success, *__ = reversed(output.decode().split("\n"))
        except ValueError:
            continue

        if success.startswith("Successfully tagged"):
            return image_id
    return None


def format_docker_build_context_help_message(
    context_root: str, context: DockerBuildContext, colors: bool
) -> str | None:
    paths_outside_context_root: list[str] = []

    def _chroot_context_paths(paths: tuple[str, str]) -> tuple[str, str]:
        """Adjust the context paths in `copy_source_vs_context_source` for `context_root`."""
        instruction_path, context_path = paths
        if not context_path:
            return paths
        dst = os.path.relpath(context_path, context_root)
        if dst.startswith("../"):
            paths_outside_context_root.append(context_path)
            return ("", "")
        if instruction_path == dst:
            return ("", "")
        return instruction_path, dst

    # Adjust context paths based on `context_root`.
    copy_source_vs_context_source: tuple[tuple[str, str], ...] = tuple(
        filter(any, map(_chroot_context_paths, context.copy_source_vs_context_source))
    )

    if not (copy_source_vs_context_source or paths_outside_context_root):
        # No issues found.
        return None

    msg = ""
    has_unsourced_copy = any(src for src, _ in copy_source_vs_context_source)
    if has_unsourced_copy:
        msg += (
            f"The {context.dockerfile} has `COPY` instructions for source files that may not have "
            f"been found in the Docker build context.\n\n"
        )

        renames = sorted(
            format_rename_suggestion(src, dst, colors=colors)
            for src, dst in copy_source_vs_context_source
            if src and dst
        )
        if renames:
            msg += (
                f"However there are possible matches. Please review the following list of "
                f"suggested renames:\n\n{bullet_list(renames)}\n\n"
            )

        unknown = sorted(src for src, dst in copy_source_vs_context_source if src and not dst)
        if unknown:
            msg += (
                f"The following files were not found in the Docker build context:\n\n"
                f"{bullet_list(unknown)}\n\n"
            )

    unreferenced = sorted(dst for src, dst in copy_source_vs_context_source if dst and not src)
    if unreferenced:
        msg += (
            f"There are files in the Docker build context that were not referenced by "
            f"any `COPY` instruction (this is not an error):\n\n{bullet_list(unreferenced, 10)}\n\n"
        )

    if paths_outside_context_root:
        unreachable = sorted({os.path.dirname(pth) for pth in paths_outside_context_root})
        context_paths = tuple(dst for src, dst in context.copy_source_vs_context_source if dst)
        new_context_root = os.path.commonpath(context_paths)
        msg += (
            "There are unreachable files in these directories, excluded from the build context "
            f"due to `context_root` being {context_root!r}:\n\n{bullet_list(unreachable, 10)}\n\n"
            f"Suggested `context_root` setting is {new_context_root!r} in order to include all "
            "files in the build context, otherwise relocate the files to be part of the current "
            f"`context_root` {context_root!r}."
        )

    return msg


def rules():
    return [
        *collect_rules(),
        UnionRule(PackageFieldSet, DockerPackageFieldSet),
    ]

pantsbuild / pants / 23616668936

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous