feat(application): enable build plans (#70)

Enable multi-artefact builds by using the application-defined build plan
resolver to create multiple provider instances. Further refactoring may
be necessary if applications such as Rockcraft and Charmcraft require
additional features.

Signed-off-by: Claudio Matsuoka <[email protected]>

craft_application/application.py

@@ -16,6 +16,7 @@
 """Main application classes for a craft-application."""
 from __future__ import annotations
 
+import argparse
 import functools
 import os
 import pathlib
@@ -30,7 +31,8 @@
 import craft_providers
 from xdg.BaseDirectory import save_cache_path  # type: ignore[import]
 
-from craft_application import commands, models
+from craft_application import commands, models, util
+from craft_application.models import BuildInfo
 
 if TYPE_CHECKING:
     from craft_application.services import service_factory
@@ -40,6 +42,15 @@
 )
 
 
+class _Dispatcher(craft_cli.Dispatcher):
+    """Application command dispatcher."""
+
+    @property
+    def parsed_args(self) -> argparse.Namespace:
+        """The map of parsed command-line arguments."""
+        return self._parsed_command_args or argparse.Namespace()
+
+
 @final
 @dataclass(frozen=True)
 class AppMetadata:
@@ -119,7 +130,7 @@ def cache_dir(self) -> str:
         # xdg types: https://github.com/python/typeshed/pull/10163
         return save_cache_path(self.app.name)  # type: ignore[no-any-return]
 
-    def _configure_services(self) -> None:
+    def _configure_services(self, build_for: str | None) -> None:
         """Configure additional keyword arguments for any service classes.
 
         Any child classes that override this must either call this directly or must
@@ -129,35 +140,52 @@ def _configure_services(self) -> None:
             "lifecycle",
             cache_dir=self.cache_dir,
             work_dir=self._work_dir,
+            build_for=build_for,
         )
 
     @functools.cached_property
     def project(self) -> models.Project:
         """Get this application's Project metadata."""
         project_file = (self._work_dir / f"{self.app.name}.yaml").resolve()
+        craft_cli.emit.debug(f"Loading project file '{project_file!s}'")
         return self.app.ProjectClass.from_yaml_file(project_file)
 
-    def run_managed(self) -> None:
+    def run_managed(self, build_for: str | None) -> None:
         """Run the application in a managed instance."""
-        craft_cli.emit.debug(f"Running {self.app.name} in a managed instance...")
-        instance_path = pathlib.PosixPath("/root/project")
-        with self.services.provider.instance(
-            self.project.effective_base, work_dir=self._work_dir
-        ) as instance:
-            try:
-                # Pyright doesn't fully understand craft_providers's CompletedProcess.
-                instance.execute_run(  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
-                    [self.app.name, *sys.argv[1:]], cwd=instance_path, check=True
-                )
-            except subprocess.CalledProcessError as exc:
-                raise craft_providers.ProviderError(
-                    f"Failed to execute {self.app.name} in instance."
-                ) from exc
+        extra_args: dict[str, Any] = {}
+
+        build_plan = self.project.get_build_plan()
+        build_plan = _filter_plan(build_plan, build_for)
+
+        if build_for:
+            extra_args["env"] = {"CRAFT_BUILD_FOR": build_for}
+
+        for build_info in build_plan:
+            craft_cli.emit.debug(
+                f"Running {self.app.name} in {build_info.build_for} instance..."
+            )
+            instance_path = pathlib.PosixPath("/root/project")
+
+            with self.services.provider.instance(
+                build_info, work_dir=self._work_dir
+            ) as instance:
+                try:
+                    # Pyright doesn't fully understand craft_providers's CompletedProcess.
+                    instance.execute_run(  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
+                        [self.app.name, *sys.argv[1:]],
+                        cwd=instance_path,
+                        check=True,
+                        **extra_args,
+                    )
+                except subprocess.CalledProcessError as exc:
+                    raise craft_providers.ProviderError(
+                        f"Failed to execute {self.app.name} in instance."
+                    ) from exc
 
     def configure(self, global_args: dict[str, Any]) -> None:
         """Configure the application using any global arguments."""
 
-    def _get_dispatcher(self) -> craft_cli.Dispatcher:
+    def _get_dispatcher(self) -> _Dispatcher:
         """Configure this application. Should be called by the run method.
 
         Side-effect: This method may exit the process.
@@ -171,7 +199,7 @@ def _get_dispatcher(self) -> craft_cli.Dispatcher:
             log_filepath=self.log_path,
         )
 
-        dispatcher = craft_cli.Dispatcher(
+        dispatcher = _Dispatcher(
             self.app.name,
             self.command_groups,
             summary=str(self.app.summary),
@@ -223,7 +251,6 @@ def _get_dispatcher(self) -> craft_cli.Dispatcher:
     def run(self) -> int:
         """Bootstrap and run the application."""
         dispatcher = self._get_dispatcher()
-        self._configure_services()
         craft_cli.emit.trace("Preparing application...")
 
         return_code = 1  # General error
@@ -237,14 +264,19 @@ def run(self) -> int:
                     }
                 ),
             )
+            build_for = getattr(dispatcher.parsed_args, "build_for", None)
+            self._configure_services(build_for)
+
             if not command.run_managed:
                 # command runs in the outer instance
                 craft_cli.emit.debug(f"Running {self.app.name} {command.name} on host")
+                if command.always_load_project:
+                    self.services.project = self.project
                 return_code = dispatcher.run() or 0
             elif not self.services.ProviderClass.is_managed():
                 # command runs in inner instance, but this is the outer instance
                 self.services.project = self.project
-                self.run_managed()
+                self.run_managed(build_for)
                 return_code = 0
             else:
                 # command runs in inner instance
@@ -280,3 +312,18 @@ def _emit_error(
             error.logpath_report = False
 
         craft_cli.emit.error(error)
+
+
+def _filter_plan(build_plan: list[BuildInfo], build_for: str | None) -> list[BuildInfo]:
+    """Filter out builds not matching build-on and build-for."""
+    host_arch = util.get_host_architecture()
+
+    plan: list[BuildInfo] = []
+    for build_info in build_plan:
+        build_on_matches = build_info.build_on == host_arch
+        build_for_matches = not build_for or build_info.build_for == build_for
+
+        if build_on_matches and build_for_matches:
+            plan.append(build_info)
+
+    return plan

craft_application/commands/base.py

@@ -33,6 +33,9 @@ class AppCommand(BaseCommand):
     run_managed: bool = False
     """Whether this command should run in managed mode."""
 
+    always_load_project: bool = False
+    """The project is also loaded in non-managed mode."""
+
     def __init__(self, config: dict[str, Any]) -> None:
         super().__init__(config)
         self._app: application.AppMetadata = config["app"]

craft_application/commands/lifecycle.py

@@ -14,6 +14,7 @@
 """Basic lifecycle commands for a Craft Application."""
 from __future__ import annotations
 
+import os
 import pathlib
 import textwrap
 from typing import TYPE_CHECKING
@@ -96,6 +97,14 @@ def fill_parser(self, parser: argparse.ArgumentParser) -> None:
             help="Shell into the environment after the step has run.",
         )
 
+        parser.add_argument(
+            "--build-for",
+            type=str,
+            metavar="arch",
+            default=os.getenv("CRAFT_BUILD_FOR"),
+            help="Set architecture to build for",
+        )
+
     @override
     def get_managed_cmd(self, parsed_args: argparse.Namespace) -> list[str]:
         """Get the command to run in managed mode.
@@ -119,10 +128,11 @@ def run(
     ) -> None:
         """Run a lifecycle step command."""
         super().run(parsed_args)
-
         step_name = step_name or self.name
-
-        self._services.lifecycle.run(step_name=step_name, part_names=parsed_args.parts)
+        self._services.lifecycle.run(
+            step_name=step_name,
+            part_names=parsed_args.parts,
+        )
 
 
 class PullCommand(_LifecycleStepCommand):
@@ -259,6 +269,7 @@ class CleanCommand(_LifecyclePartsCommand):
         remove the packing environment.
         """
     )
+    always_load_project = True
 
     @override
     def run(self, parsed_args: argparse.Namespace) -> None:

craft_application/models/__init__.py

@@ -24,11 +24,12 @@
     VersionStr,
 )
 from craft_application.models.metadata import BaseMetadata
-from craft_application.models.project import Project
+from craft_application.models.project import BuildInfo, Project
 
 
 __all__ = [
     "BaseMetadata",
+    "BuildInfo",
     "CraftBaseConfig",
     "CraftBaseModel",
     "Project",

craft_application/models/project.py

@@ -17,9 +17,11 @@
 
 This defines the structure of the input file (e.g. snapcraft.yaml)
 """
-from typing import Any, Dict, Optional, Union
+import dataclasses
+from typing import Any, Dict, List, Optional, Union
 
 import craft_parts
+import craft_providers.bases
 import pydantic
 from pydantic import AnyUrl
 
@@ -33,6 +35,15 @@
 )
 
 
+@dataclasses.dataclass
+class BuildInfo:
+    """Platform build information."""
+
+    build_on: str
+    build_for: str
+    base: craft_providers.bases.BaseName
+
+
 class Project(CraftBaseModel):
     """Craft Application project definition."""
 
@@ -67,3 +78,9 @@ def effective_base(self) -> Any:  # noqa: ANN401 app specific classes can improv
         if self.base is not None:
             return self.base
         raise RuntimeError("Could not determine effective base")
+
+    def get_build_plan(self) -> List[BuildInfo]:
+        """Obtain the list of architectures and bases from the project file."""
+        raise NotImplementedError(
+            f"{self.__class__.__name__!s} must implement get_build_plan"
+        )

craft_application/services/lifecycle.py

@@ -24,6 +24,7 @@
 
 from craft_application import errors
 from craft_application.services import base
+from craft_application.util import convert_architecture_deb_to_platform
 
 if TYPE_CHECKING:  # pragma: no cover
     from pathlib import Path
@@ -102,22 +103,26 @@ class LifecycleService(base.BaseService):
     :param app: An AppMetadata object containing metadata about the application.
     :param project: The Project object that describes this project.
     :param work_dir: The working directory for parts processing.
+    :param cache_dir: The cache directory for parts processing.
+    :param build_for: The architecture or platform we are building for.
     :param lifecycle_kwargs: Additional keyword arguments are passed through to the
         LifecycleManager on initialisation.
     """
 
-    def __init__(
+    def __init__(  # noqa: PLR0913 (too many arguments)
         self,
         app: AppMetadata,
         project: Project,
         *,
         work_dir: Path | str,
         cache_dir: Path | str,
+        build_for: str,
         **lifecycle_kwargs: Any,  # noqa: ANN401 - eventually used in an Any
     ) -> None:
         super().__init__(app, project)
         self._work_dir = work_dir
         self._cache_dir = cache_dir
+        self._build_for = build_for
         self._manager_kwargs = lifecycle_kwargs
         self._lcm = self._init_lifecycle_manager()
 
@@ -133,6 +138,7 @@ def _init_lifecycle_manager(self) -> LifecycleManager:
             return LifecycleManager(
                 {"parts": self._project.parts},
                 application_name=self._app.name,
+                arch=convert_architecture_deb_to_platform(self._build_for),
                 cache_dir=self._cache_dir,
                 work_dir=self._work_dir,
                 ignore_local_sources=self._app.source_ignore_patterns,
@@ -183,7 +189,8 @@ def clean(self, part_names: list[str] | None = None) -> None:
     def __repr__(self) -> str:
         work_dir = self._work_dir
         cache_dir = self._cache_dir
+        build_for = self._build_for
         return (
             f"{self.__class__.__name__}({self._app!r}, {self._project!r}, "
-            f"{work_dir=}, {cache_dir=}, **{self._manager_kwargs!r})"
+            f"{work_dir=}, {cache_dir=}, {build_for=}, **{self._manager_kwargs!r})"
         )

craft_application/services/provider.py

@@ -71,7 +71,7 @@ def is_managed(cls) -> bool:
     @contextlib.contextmanager
     def instance(
         self,
-        base_name: bases.BaseName | tuple[str, str],
+        build_info: models.BuildInfo,
         *,
         work_dir: pathlib.Path,
         allow_unstable: bool = True,
@@ -84,9 +84,13 @@ def instance(
         :param allow_unstable: Whether to allow the use of unstable images.
         :returns: a context manager of the provider instance.
         """
-        emit.debug("Preparing managed instance")
         work_dir_inode = work_dir.stat().st_ino
-        instance_name = f"{self._app.name}-{self._project.name}-{work_dir_inode}"
+        instance_name = (
+            f"{self._app.name}-{self._project.name}-on-{build_info.build_on}-"
+            f"for-{build_info.build_for}-{work_dir_inode}"
+        )
+        emit.debug("Preparing managed instance {instance_name!r}")
+        base_name = build_info.base
         base = self.get_base(base_name, instance_name=instance_name, **kwargs)
         provider = self.get_provider()
 

craft_application/util/__init__.py

@@ -16,7 +16,13 @@
 """Utilities for craft-application."""
 
 from craft_application.util.yaml import safe_yaml_load
+from craft_application.util.platforms import (
+    get_host_architecture,
+    convert_architecture_deb_to_platform,
+)
 
 __all__ = [
     "safe_yaml_load",
+    "get_host_architecture",
+    "convert_architecture_deb_to_platform",
 ]