fix: finalize cmd: skip processing blobs of OTA image metadata files/non-image payloads (#84)

This PR fixes an edge condition causing OTA failed due to blob missing caused by finalize cmd's blob storage optimizing when blobs come from OTA image metadata files(like sys_config.yaml) or from otaclient release packages also presented in the OTA image payload.

This PR introduces a mechanism to collect blobs' digests that belongs to OTA image metadata files(like sys_config.yaml) or otaclient packages, and skip these digests when optimizing the blobs storage.

Other minor fix:
1. Dockerfile: fix alpha/beta/rc build version file generation.

With this PR, ota-image-builder will bump to version v0.9.0.

Author: Bodong-Yang

docker/builder_release/Dockerfile

@@ -6,17 +6,11 @@ FROM ${BUILD_BASE} AS app_build
 
 ARG PYINSTALLER_VER=6.16.0
 
-RUN mkdir /build
-# minimum required files for building app
-COPY ./src /build/src
-COPY ./pyproject.toml ./uv.lock ./README.md /build/
-
+COPY . /build
 COPY --from=ghcr.io/astral-sh/uv:0.9.4 /uv /uvx /bin/
 
 WORKDIR /build
-# NOTE: mount .git for hatch-vcs to determine the version
-RUN --mount=type=bind,source=./.git,target=/build/.git,ro \
-    set -eux; \
+RUN set -eux; \
     apt-get update -qq; \
     apt-get install -y -qq --no-install-recommends \
         git python3 python3-setuptools; \

docker/e2e_test_base/Dockerfile

@@ -10,9 +10,7 @@ FROM ${SYS_IMG} AS sys_img
 
 FROM ${UBUNTU_BASE} AS app_builder
 
-# copy minimum required files for build
-COPY ./pyproject.toml ./.python-version ./uv.lock /project/
-COPY ./src /project/src
+COPY . /project
 
 ARG UV_VERSION
 

src/ota_image_builder/cmds/finalize.py

@@ -22,6 +22,8 @@
 
 from ota_image_libs.common import tmp_fname
 from ota_image_libs.v1.image_index.utils import ImageIndexHelper
+from ota_image_libs.v1.image_manifest.schema import ImageManifest
+from ota_image_libs.v1.otaclient_package.schema import OTAClientPackageManifest
 from ota_image_libs.v1.resource_table.schema import (
     ZstdCompressedResourceTableDescriptor,
 )
@@ -89,6 +91,52 @@ def finalize_cmd_args(
     finalize_cmd_args.set_defaults(handler=finalize_cmd)
 
 
+def _collect_protected_resources_digest(_index_helper: ImageIndexHelper) -> set[bytes]:
+    """Scan through OTA image, collect blob digests that don't belong to any system image.
+
+    When optimizing the blob storage, we MUST skip processing these digests.
+    The blobs that don't belong any image payload:
+    1. image_payload: sys_config and file_table files.
+    2. otaclient_release: manifest.json and release packages.
+    3. resource_table itself.
+
+    NOTE(20251219): an example case is when the system image is dev build, and contains
+                    the pilot-auto source code within the built system image.
+                    This will result in blob of sys_config file is also part of the system image,
+                    thus being processed during blob storage optimization, and the original blob
+                    being removed.
+    """
+    _res: set[bytes] = set()
+    _resource_dir = _index_helper.image_resource_dir
+    for manifest_descriptor in _index_helper.image_index.manifests:
+        _res.add(manifest_descriptor.digest.digest)
+        if isinstance(manifest_descriptor, ImageManifest.Descriptor):
+            _manifest = manifest_descriptor.load_metafile_from_resource_dir(
+                _resource_dir
+            )
+
+            for _file_table_descriptor in _manifest.layers:
+                _res.add(_file_table_descriptor.digest.digest)
+
+            _image_config_descriptor = _manifest.config
+            _res.add(_image_config_descriptor.digest.digest)
+
+            _image_config = _image_config_descriptor.load_metafile_from_resource_dir(
+                _resource_dir
+            )
+            if _sys_config_descriptor := _image_config.sys_config:
+                _res.add(_sys_config_descriptor.digest.digest)
+            _res.add(_image_config.file_table.digest.digest)
+        elif isinstance(manifest_descriptor, OTAClientPackageManifest.Descriptor):
+            _manifest = manifest_descriptor.load_metafile_from_resource_dir(
+                _resource_dir
+            )
+            _res.add(_manifest.config.digest.digest)
+            for _payload in _manifest.layers:
+                _res.add(_payload.digest.digest)
+    return _res
+
+
 def finalize_cmd(args: Namespace) -> None:
     logger.debug(f"calling {finalize_cmd.__name__} with {args}")
     image_root = Path(args.image_root)
@@ -98,6 +146,10 @@ def finalize_cmd(args: Namespace) -> None:
     index_helper = ImageIndexHelper(image_root)
     logger.info(f"Finalize and optimize OTA image at {image_root} ...")
     logger.info("Optimizing the blob storage of the OTA image ...")
+    protected_resources = _collect_protected_resources_digest(index_helper)
+    logger.debug(
+        f"Skip the protected resources: {[d.hex() for d in protected_resources]}"
+    )
 
     resource_dir = index_helper.image_resource_dir
     _old_rstable_descriptor = index_helper.image_index.image_resource_table
@@ -123,7 +175,9 @@ def finalize_cmd(args: Namespace) -> None:
         if not args.o_skip_bundle:
             logger.info("Apply bundle filter to the blob storage ...")
             BundleFilterProcesser(
-                resource_dir=resource_dir, rst_dbf=_working_rstable
+                resource_dir=resource_dir,
+                rst_dbf=_working_rstable,
+                protected_resources=protected_resources,
             ).process()
             logger.info(
                 f"Finish applying bundle filter: time cost: {int(time.time() - start_time)}s"
@@ -135,7 +189,9 @@ def finalize_cmd(args: Namespace) -> None:
             logger.info("Apply compression filter to the blob storage ...")
             _start_time = time.time()
             CompressionFilterProcesser(
-                resource_dir=resource_dir, rst_dbf=_working_rstable
+                resource_dir=resource_dir,
+                rst_dbf=_working_rstable,
+                protected_resources=protected_resources,
             ).process()
             logger.info(
                 f"Finish applying compression filter: time cost: {int(time.time() - _start_time)}s"
@@ -149,7 +205,9 @@ def finalize_cmd(args: Namespace) -> None:
             logger.info("Apply slice filter to the blob storage ...")
             _start_time = time.time()
             SliceFilterProcesser(
-                resource_dir=resource_dir, rst_dbf=_working_rstable
+                resource_dir=resource_dir,
+                rst_dbf=_working_rstable,
+                protected_resources=protected_resources,
             ).process()
             logger.info(
                 f"Finish applying slice filter: time cost: {int(time.time() - _start_time)}s"

src/ota_image_builder/v1/_image_config.py

@@ -16,10 +16,8 @@
 
 import logging
 from datetime import datetime
-from pathlib import Path
 from typing import Any
 
-import yaml
 from ota_image_libs.common import AliasEnabledModel
 from ota_image_libs.v1.annotation_keys import (
     OS,
@@ -50,23 +48,6 @@ class AddImageConfigAnnotations(AliasEnabledModel):
     os_version: str | None = Field(alias=OS_VERSION, default=None)
 
 
-def add_sys_config(sys_cfg_fpath: Path, resource_dir: Path) -> SysConfig.Descriptor:
-    try:
-        _raw = yaml.safe_load(sys_cfg_fpath.read_text())
-        SysConfig.model_validate(_raw)
-    except Exception as e:
-        raise ValueError(f"invalid sys_config file {sys_cfg_fpath}") from e
-    return SysConfig.Descriptor.add_file_to_resource_dir(sys_cfg_fpath, resource_dir)
-
-
-def add_file_table(
-    file_table_fpath: Path, resource_dir: Path
-) -> ZstdCompressedFileTableDescriptor:
-    return ZstdCompressedFileTableDescriptor.add_file_to_resource_dir(
-        file_table_fpath, resource_dir, remove_origin=True
-    )
-
-
 def compose_image_config(
     *,
     file_table_descriptor: ZstdCompressedFileTableDescriptor,

src/ota_image_builder/v1/_resource_process/_bundle_filter.py

@@ -69,16 +69,19 @@ class BundleCompressedResult(NamedTuple):
     compressed_size: int
 
 
-def _batch_entries(
+def _batch_entries_with_filter(
     entries_to_bundle_gen: Generator[EntryToBeBundled],
     *,
     expected_bundle_size: int,
     min_bundle_ratio: float = MINIMUM_BUNDLE_SIZE_RATIO,
+    excluded_resources: set[bytes],
 ) -> Generator[tuple[int, list[EntryToBeBundled]]]:
     _batch = []
     _this_batch_size = 0
     for _entry in entries_to_bundle_gen:
-        _, _, _entry_size = _entry
+        _, _digest, _entry_size = _entry
+        if _digest in excluded_resources:
+            continue
 
         _batch.append(_entry)
         _this_batch_size += _entry_size
@@ -200,7 +203,9 @@ def __init__(
         bundle_upper_bound: int = cfg.BUNDLE_UPPER_THRESHOULD,
         bundle_blob_size: int = cfg.BUNDLE_SIZE,
         bundle_compressed_max_sum: int = cfg.BUNDLES_COMPRESSED_MAXIMUM_SUM,
+        protected_resources: set[bytes],
     ) -> None:
+        self._protected_resources = protected_resources
         self._resource_dir = resource_dir
         self._db_helper = ResourceTableDBHelper(rst_dbf)
         self._lower_bound = bundle_lower_bound
@@ -232,8 +237,10 @@ def process(self):
                 _row_factory=sqlite3.Row,
             ) # type: ignore[assignment]
             # fmt: on
-            batch_gen = _batch_entries(
-                entries_to_bundle_gen, expected_bundle_size=self._bundle_blob_size
+            batch_gen = _batch_entries_with_filter(
+                entries_to_bundle_gen,
+                expected_bundle_size=self._bundle_blob_size,
+                excluded_resources=self._protected_resources,
             )
 
             cctx = zstandard.ZstdCompressor(level=cfg.BUNDLE_ZSTD_COMPRESSION_LEVEL)

src/ota_image_builder/v1/_resource_process/_compression_filter.py

@@ -63,7 +63,9 @@ def __init__(
         read_size: int = cfg.READ_SIZE,
         worker_threads: int = cfg.COMPRESSION_RESOURCE_SCAN_WORKER_THREADS,
         concurrent_jobs: int = cfg.COMPRESSION_MAX_CONCURRENT,
+        protected_resources: set[bytes],
     ) -> None:
+        self._protected_resources = protected_resources
         self._read_size = read_size
         self._resource_dir = resource_dir
         self._db_helper = ResourceTableDBHelper(rst_dbf)
@@ -157,6 +159,9 @@ def process(self) -> None:
                 for _raw_row in rs_orm.orm_select_entries(
                     _stmt=_stmt, _row_factory=sqlite3.Row
                 ):
+                    if _raw_row[1] in self._protected_resources:
+                        continue
+
                     origin_size += _raw_row[-1]
                     submit_with_se(
                         self._process_one_entry_at_thread,

src/ota_image_builder/v1/_resource_process/_slice_filter.py

@@ -119,7 +119,9 @@ def __init__(
         worker_threads: int = cfg.WORKER_THREADS,
         concurrent_tasks: int = cfg.SLICE_CONCURRENT_TASKS,
         db_update_batch_size: int = cfg.SLICE_UPDATE_BATCH_SIZE,
+        protected_resources: set[bytes],
     ) -> None:
+        self._protected_resources = protected_resources
         self._update_batch = db_update_batch_size
         self._worker_threads = worker_threads
         self._se = Semaphore(concurrent_tasks)
@@ -214,6 +216,9 @@ def process(self):
                     _row_factory=sqlite3.Row,
                 ):
                     resource_id, entry_digest, entry_size = _row
+                    if entry_digest in self._protected_resources:
+                        continue
+
                     sliced_count += 1
                     sliced_size += entry_size