Refactor image target metadata handling to allow multiple metadata attachments

posit-bakery/posit_bakery/cli/ci.py

@@ -172,44 +172,28 @@ def merge(
     log.info(f"Reading targets from {', '.join(f.name for f in metadata_file)}")
 
     files_ok = True
-    image_digests: dict[str, list[str]] = {}
+    loaded_targets: list[str] = []
     for file in metadata_file:
-        if not file.is_file():
-            log.error(f"Metadata file '{file}' does not exist")
+        try:
+            loaded_targets.extend(config.load_build_metadata_from_file(file))
+        except Exception as e:
+            log.error(f"Failed to load metadata from file '{file}'")
+            log.error(str(e))
             files_ok = False
-            continue
-        with open(file, "r") as f:
-            try:
-                data = json.load(f)
-            except json.JSONDecodeError as e:
-                log.error(f"Metadata file '{file}' is not valid JSON: {str(e)}")
-                files_ok = False
-                continue
-        for uid, metadata in data.items():
-            image_name = metadata.get("image.name")
-            if not image_name:
-                log.error(f"Metadata file '{file}' is missing 'image.name' for image UID '{uid}'")
-                files_ok = False
-                continue
-            digest = metadata.get("containerimage.digest")
-            if not digest:
-                log.error(f"Metadata file '{file}' is missing 'containerimage.digest' for image UID '{uid}'")
-                files_ok = False
-                continue
-            image_digests.setdefault(uid, []).append(f"{image_name}@{digest}")
+    loaded_targets = list(set(loaded_targets))  # Deduplicate targets in case of overlap across files
 
     if not files_ok:
         log.error("One or more metadata files are invalid, aborting merge.")
         raise typer.Exit(code=1)
 
-    log.info(f"Found {len(image_digests.keys())} targets")
-    log.debug(json.dumps(image_digests, indent=2, sort_keys=True))
+    log.info(f"Found {len(loaded_targets)} targets")
+    log.debug(", ".join(loaded_targets))
 
-    for uid, sources in image_digests.items():
+    for uid in loaded_targets:
         target = config.get_image_target_by_uid(uid)
-        log.info(f"Merging {len(sources)} sources for image UID '{uid}'")
+        log.info(f"Merging sources for image UID '{uid}'")
         try:
-            manifest = target.merge(sources=sources, dry_run=dry_run)
+            manifest = target.merge(dry_run=dry_run)
             stdout_console.print_json(manifest.model_dump_json(indent=2, exclude_unset=True, exclude_none=True))
         except DockerException as e:
             log.error(f"Error merging sources for UID '{uid}'")

posit-bakery/posit_bakery/cli/run.py

@@ -1,5 +1,4 @@
 import logging
-import platform
 import re
 from enum import Enum
 from pathlib import Path
@@ -12,6 +11,7 @@
 from posit_bakery.config.config import BakeryConfigFilter, BakerySettings
 from posit_bakery.const import DevVersionInclusionEnum, MatrixVersionInclusionEnum
 from posit_bakery.log import stderr_console
+from posit_bakery.settings import SETTINGS
 from posit_bakery.util import auto_path
 
 log = logging.getLogger(__name__)
@@ -75,7 +75,7 @@ def dgoss(
     image_platform: Annotated[
         Optional[str],
         typer.Option(
-            show_default=platform.machine(),  # TODO: improve output to match docker platform format
+            show_default=SETTINGS.get_host_architecture(),
             help="Filters which image build platform to run tests for, e.g. 'linux/amd64'. Image test targets "
             "incompatible with the given platform(s) will be skipped. Requires a compatible goss binary.",
             rich_help_panel=RichHelpPanelEnum.FILTERS,
@@ -121,14 +121,8 @@ def dgoss(
     `DGOSS_BIN` environment variables if not present in the system PATH.
     """
     # Autoselect host architecture platform if not specified.
-    if image_platform is None:
-        machine = platform.machine()
-        arch_map = {
-            "x86_64": "amd64",
-            "aarch64": "arm64",
-        }
-        arch = arch_map.get(machine, "amd64")
-        image_platform = f"linux/{arch}"
+    image_platform = image_platform or SETTINGS.architecture
+    image_platform = f"linux/{image_platform}"
 
     settings = BakerySettings(
         filter=BakeryConfigFilter(

posit-bakery/posit_bakery/config/config.py

@@ -35,6 +35,7 @@
 from posit_bakery.image.bake.bake import BakePlan
 from posit_bakery.image.goss.dgoss import DGossSuite
 from posit_bakery.image.goss.report import GossJsonReportCollection
+from posit_bakery.image.image_metadata import MetadataFile
 from posit_bakery.image.image_target import ImageTarget, ImageBuildStrategy, ImageTargetSettings
 from posit_bakery.registry_management import ghcr
 
@@ -850,20 +851,27 @@ def _merge_sequential_build_metadata_files(self) -> dict[str, Any]:
         """
         merged_metadata: dict[str, dict[str, Any]] = {}
         for target in self.targets:
-            if target.metadata_file is not None:
-                merged_metadata[target.uid] = target.metadata_file.metadata.model_dump(exclude_none=True, by_alias=True)
+            for build_metadata in target.build_metadata:
+                merged_metadata[target.uid] = build_metadata.model_dump(exclude_none=True, by_alias=True)
+
         return merged_metadata
 
-    def load_build_metadata_from_file(self, metadata_file: Path):
+    def load_build_metadata_from_file(self, metadata_file: Path) -> list[str]:
         """Loads build metadata from a given metadata file.
 
         :param metadata_file: Path to the metadata file to load.
-        :return: A dictionary containing the loaded metadata.
+        :return: A list of targets loaded.
         """
-        if not metadata_file.is_file():
-            raise FileNotFoundError(f"Metadata file '{str(metadata_file)}' does not exist.")
+        metadata_file = MetadataFile.load(metadata_file)
+
+        targets_loaded = []
         for target in self.targets:
-            target.load_build_metadata_from_file(metadata_file)
+            result = target.load_build_metadata_from_file(metadata_file)
+            if result is not None:
+                targets_loaded.append(target.uid)
+                log.info(f"Loaded build metadata for target '{target}' from file '{metadata_file.filepath}'.")
+
+        return targets_loaded
 
     def bake_plan_targets(self) -> str:
         """Generates a bake plan JSON string for the image targets defined in the config."""

posit-bakery/posit_bakery/image/goss/dgoss.py

@@ -158,7 +158,10 @@ def command(self) -> list[str]:
         if self.runtime_options:
             # TODO: We may want to validate this to ensure options are not duplicated.
             cmd.extend(self.runtime_options.split())
-        cmd.append(self.image_target.ref)
+        if self.platform:
+            cmd.append(self.image_target.ref(self.platform))
+        else:
+            cmd.append(self.image_target.ref())
         cmd.extend(self.image_command.split())
 
         return cmd

posit-bakery/posit_bakery/image/image_metadata.py

@@ -1,8 +1,12 @@
+import datetime
 import json
+import logging
 from pathlib import Path
 from typing import Annotated, Self
 
-from pydantic import ConfigDict, BaseModel, Field, model_validator
+from pydantic import ConfigDict, BaseModel, Field, RootModel
+
+log = logging.getLogger(__name__)
 
 
 class ImageToolsInspectionPlatformMetadata(BaseModel):
@@ -74,6 +78,49 @@ class BuildMetadataContainerImageDescriptor(BaseModel):
     ]
 
 
+class BuildMetadataBuildProvenanceMaterial(BaseModel):
+    """Representation of a material used in the build process."""
+
+    model_config = ConfigDict(extra="allow")
+
+    uri: Annotated[str | None, Field(description="The URI of the material.", default=None)]
+    digest: Annotated[dict[str, str] | None, Field(description="The digest of the material.", default=None)]
+
+
+class BuildMetadataBuildProvenanceInvocation(BaseModel):
+    """Representation of the invocation of the build process."""
+
+    model_config = ConfigDict(extra="allow")
+
+    config_source: Annotated[
+        dict,
+        Field(
+            description="The configuration source of the build invocation.", alias="configSource", default_factory=dict
+        ),
+    ]
+    parameters: Annotated[dict, Field(description="The parameters of the build invocation.", default_factory=dict)]
+    environment: Annotated[dict, Field(description="The environment of the build invocation.", default_factory=dict)]
+
+
+class BuildMetadataBuildProvenance(BaseModel):
+    """Representation of build provenance in build metadata."""
+
+    model_config = ConfigDict(extra="allow")
+
+    builder: Annotated[dict, Field(description="The builder used to build the image.", default_factory=dict)]
+    build_type: Annotated[
+        str | None, Field(description="The type of build performed.", alias="buildType", default=None)
+    ]
+    materials: Annotated[
+        list[BuildMetadataBuildProvenanceMaterial],
+        Field(description="The materials used in the build process.", default_factory=list),
+    ]
+    invocation: Annotated[
+        BuildMetadataBuildProvenanceInvocation | None,
+        Field(description="The invocation of the build process.", default=None),
+    ]
+
+
 class BuildMetadata(BaseModel):
     """Representation of build metadata produced by Docker builds."""
 
@@ -93,6 +140,10 @@ class BuildMetadata(BaseModel):
             default=None,
         ),
     ]
+    build_provenance: Annotated[
+        BuildMetadataBuildProvenance | None,
+        Field(description="The build provenance of the built image.", alias="buildx.build.provenance", default=None),
+    ]
 
     @property
     def image_tags(self) -> list[str]:
@@ -110,24 +161,82 @@ def image_ref(self) -> str | None:
             return primary_tag
         return None
 
+    @property
+    def created_at(self) -> datetime.datetime:
+        """Returns the creation timestamp of the built image if available."""
+        if self.container_image_descriptor and self.container_image_descriptor.annotations:
+            dt_str = self.container_image_descriptor.annotations.get("org.opencontainers.image.created")
+            if dt_str:
+                try:
+                    return datetime.datetime.fromisoformat(dt_str)
+                except ValueError:
+                    pass
+        if self.build_provenance:
+            # If the creation timestamp is not available in the annotations, we can use the build start time from
+            # labels.
+            if self.build_provenance.invocation and self.build_provenance.invocation.parameters:
+                start_time_str = self.build_provenance.invocation.parameters.get("args", {}).get(
+                    "label:org.opencontainers.image.created"
+                )
+                if start_time_str:
+                    try:
+                        return datetime.datetime.fromisoformat(start_time_str)
+                    except ValueError:
+                        pass
+        log.debug("Creation timestamp not found in metadata, defaulting to current time.")
+        return datetime.datetime.now()
+
+    @property
+    def platform(self) -> str | None:
+        """Returns the platform of the built image if available."""
+        # First, check the platform information in the container image descriptor, as it is more likely to be accurate
+        # for multi-platform builds.
+        if self.container_image_descriptor and self.container_image_descriptor.platform:
+            platform = self.container_image_descriptor.platform
+            if platform.os and platform.architecture:
+                return f"{platform.os}/{platform.architecture}"
+        # If platform information is not available in the container image descriptor, we can check the build provenance
+        # invocation environment as that should match the image platform when unspecified.
+        if self.build_provenance and self.build_provenance.invocation and self.build_provenance.invocation.environment:
+            platform = self.build_provenance.invocation.environment.get("platform")
+            if platform:
+                return platform
+
+        return None
+
+
+class BuildMetadataMap(RootModel[dict[str, BuildMetadata]]):
+    """Representation of a mapping from target UIDs to build metadata."""
+
 
 class MetadataFile(BaseModel):
-    target_uid: Annotated[str, Field(description="The target UID associated with the metadata.")]
     filepath: Annotated[Path | None, Field(description="The path to the metadata file.", default=None)]
-    metadata: Annotated[BuildMetadata | None, Field(description="The build metadata.", default=None)]
-
-    @model_validator(mode="after")
-    def validate_metadata(self) -> Self:
-        """Validates that metadata is provided."""
-        if self.metadata is None:
-            if self.filepath is None or not self.filepath.is_file():
-                raise ValueError("Either filepath or metadata must be provided.")
-            with open(self.filepath, "r") as f:
-                content = json.load(f)
-                if not isinstance(content, dict):
-                    raise ValueError("The metadata file does not contain a valid JSON object.")
-                if self.target_uid in content.keys():
-                    content = content[self.target_uid]
-                self.metadata = BuildMetadata.model_validate(content, by_alias=True)
-
-        return self
+    metadata_map: Annotated[BuildMetadataMap, Field(description="The build metadata.")]
+
+    def __repr__(self):
+        """Returns a string representation of the metadata file."""
+        return f"MetadataFile(filepath={self.filepath.absolute()}, metadata_map={self.metadata_map})"
+
+    @classmethod
+    def load(cls, filepath: Path) -> Self:
+        """Creates a MetadataFile instance from a JSON file."""
+        if not filepath.is_file():
+            raise FileNotFoundError(f"Metadata file '{str(filepath)}' does not exist.")
+
+        with open(filepath, "r") as f:
+            data = json.load(f)
+
+        metadata_map = BuildMetadataMap.model_validate(data)
+        return cls(filepath=filepath, metadata_map=metadata_map)
+
+    @classmethod
+    def loads(cls, json_str: str) -> Self:
+        """Creates a MetadataFile instance from a JSON string."""
+        data = json.loads(json_str)
+
+        metadata_map = BuildMetadataMap.model_validate(data)
+        return cls(metadata_map=metadata_map)
+
+    def get_target_metadata_by_uid(self, target_uid: str) -> BuildMetadata | None:
+        """Returns the build metadata associated with a given target UID."""
+        return self.metadata_map.root.get(target_uid)