API reference

mender_docker_lifecycle_helper.artifact

Classes:

• LifecycleHelperArtifact –

  Representation of the artifact on which the helper is operating. Provides the gen_artifact_file method to perform necessary processing and preparation for creating a Mender artifact file.

LifecycleHelperArtifact

LifecycleHelperArtifact(context: LifecycleHelperContext, name: str, artifact_metadata: ArtifactMetadata, filename: Path)

Representation of the artifact on which the helper is operating. Provides the gen_artifact_file method to perform necessary processing and preparation for creating a Mender artifact file.

Construct a LifecycleHelperArtifact object.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• name

  (str) –

  The name of the artifact as recognized by Mender.

• artifact_metadata

  (ArtifactMetadata) –

  The metadata of the artifact for caching.

• filename

  (Path) –

  The file at which to generate the artifact.

Returned by:

• API reference  helper  LifecycleHelper  create_artifact

Used by:

• API reference
  •  helper  LifecycleHelper  deploy_artifact
  •  mender_server  upload_artifact

Methods:

• call_mender_artifact –

  Execute the mender-artifact executable with the supplied arguments.

• gen_artifact_file –

  Prepare the artifact by generating necessary files and processing via the mender-artifact tool.

• gen_artifact_services –

  Determine the services metadata for the current artifact, including reading the image hashes from remote images when required. To establish this list, any provided service-file image archives are extracted and read.

• prep_artifact_dir –

  Prepare artifact directory with necessary files for artifact generation, such as extracted service file images and any other necessary files.

• prep_delta_image –

  Prepare a delta image file, which may be an empty delta file if the image ref and hash for the service match those in the previous artifact metadata, or a real delta.

• prep_image –

  Prepare the required directory and files for the specified image in the specified artifact prep directory.

• prep_images –

  Prepare the archive file for the specified images in the Mender artifact format, downloading and/or diffing images as necessary based on the image refs and hashes.

• prep_manifests –

  Prepare the manifests archive file in the Mender artifact format.

Source code in src/mender_docker_lifecycle_helper/artifact.py

def __init__(
    self,
    context: LifecycleHelperContext,
    name: str,
    artifact_metadata: ArtifactMetadata,
    filename: Path,
):
    """
    Construct a LifecycleHelperArtifact object.

    :param context: The context of the lifecycle helper execution.
    :param name: The name of the artifact as recognized by Mender.
    :param artifact_metadata: The metadata of the artifact for caching.
    :param filename: The file at which to generate the artifact.
    """
    self.context = context
    self.name = name
    self.filename = filename
    self.artifact_metadata = artifact_metadata

    self.context.logger.debug(
        f"Creating artifact named {name} with metadata {artifact_metadata.to_dict()}"
    )

    self.depends = (
        [
            f"rootfs-image.{context.manifest_name}.version:{context.previous_artifact_metadata.version}"
        ]
        if context.delta
        else []
    )
    self.image_ids = [
        service_spec["image"]["hash"]
        for service_spec in artifact_metadata.services.values()
    ]
    self.image_ids.sort()  # Sort only for determinism

call_mender_artifact staticmethod

call_mender_artifact(context: LifecycleHelperContext, arg_list: list[str]) -> str

Execute the mender-artifact executable with the supplied arguments.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• arg_list

  (list[str]) –

  The arguments to supply to the mender-artifact executable.

Returns:

• str –

  The stdout output of the mender-artifact execution.

Raises:

• RuntimeError –

  If the mender-artifact execution fails.

Source code in src/mender_docker_lifecycle_helper/artifact.py

@staticmethod
def call_mender_artifact(
    context: LifecycleHelperContext, arg_list: list[str]
) -> str:
    """
    Execute the mender-artifact executable with the supplied arguments.

    :param context: The context of the lifecycle helper execution.
    :param arg_list: The arguments to supply to the mender-artifact executable.
    :raises RuntimeError: If the mender-artifact execution fails.
    :return: The stdout output of the mender-artifact execution.
    """
    rendered_args = "\n  ".join(arg_list)
    context.logger.debug(f"Calling mender-artifact with args: \n  {rendered_args}")
    try:
        result = subprocess.run(
            # Join/split to handle arg items with and without spaces
            ["mender-artifact", *(" ".join(arg_list).split(" "))],
            capture_output=True,
            text=True,
            check=True,
        )
        context.logger.debug(f"mender-artifact output: {result.stdout}")
        return result.stdout
    except subprocess.CalledProcessError as e:
        raise RuntimeError(
            f"mender-artifact \n  {rendered_args} failed: {e.stdout} {e.stderr}"
        )

gen_artifact_file

gen_artifact_file() -> None

Prepare the artifact by generating necessary files and processing via the mender-artifact tool.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/artifact.py

def gen_artifact_file(self) -> None:
    """
    Prepare the artifact by generating necessary files and processing via the mender-artifact tool.

    :return: None
    """
    artifact_prep_dir = self.context.temp_dir / "artifact_prep"
    images_archive_filename = "images.tar.gz"
    manifests_archive_filename = "manifests.tar.gz"
    metadata_filename = "metadata.json"

    artifact_prep_dir.mkdir()
    self.prep_artifact_dir(
        artifact_prep_dir,
        images_archive_filename,
        manifests_archive_filename,
        metadata_filename,
    )
    self.call_mender_artifact(
        self.context,
        [
            "write",
            "module-image",
            "--type app",
            f"--device-type {self.context.device_type}",
            f"--output-path {self.filename}",
            f"--artifact-name {self.name}",
            f"--file {artifact_prep_dir / images_archive_filename}",
            f"--file {artifact_prep_dir / manifests_archive_filename}",
            f"--meta-data {artifact_prep_dir / metadata_filename}",
            f"--software-name {self.context.manifest_name}",
            f"--software-version {self.artifact_metadata.version}",
        ]
        + [f"--depends {depend}" for depend in self.depends],
    )
    shutil.rmtree(artifact_prep_dir)

gen_artifact_services staticmethod

gen_artifact_services(context: LifecycleHelperContext) -> dict[str:(dict[str:(dict[str, str])])]

Determine the services metadata for the current artifact, including reading the image hashes from remote images when required. To establish this list, any provided service-file image archives are extracted and read.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

Returns:

• dict[str:(dict[str:(dict[str, str])])] –

  The services to be included in the current artifact, in the following dict structure: { serviceName: { image: { ref: str, hash: str } } }

Raises:

• ManifestContentException –

  If service name in args not found in manifest.

Source code in src/mender_docker_lifecycle_helper/artifact.py

@staticmethod
def gen_artifact_services(
    context: LifecycleHelperContext,
) -> dict[str : dict[str : dict[str, str]]]:
    """
    Determine the services metadata for the current artifact, including reading the image hashes from remote images when required. To establish this list, any provided service-file image archives are extracted and read.

    :param context: The context of the lifecycle helper execution.
    :raises ManifestContentException: If service name in args not found in manifest.
    :return: The services to be included in the current artifact, in the following dict structure:
        {
            serviceName: {
                image: {
                    ref: str,
                    hash: str
                }
            }
        }
    """

    context.logger.info("Determining services for the artifact.")
    artifact_services = {}
    for service_name, service_spec in context.manifest["services"].items():
        artifact_service = artifact_services[service_name] = {}
        artifact_service_image = artifact_service["image"] = {}
        # The metadata from each specified service file is used in the artifact, taking precedence over any specified service image override
        if service_name in context.service_files:
            if service_name in context.service_images:
                context.logger.warning(
                    f"Service {service_name} has both a service file and an image override specified. Ignoring the image override."
                )

            service_filename = context.service_files[service_name]
            service_image = context.image_cache.extract_cache_file(service_filename)
            # Both the ref and hash are set for services with a provided service file
            context.logger.debug(
                f"Overriding {service_name} image details from file {service_filename} to ref {service_image['ref']} and hash {service_image['hash']}."
            )
            artifact_service_image["ref"] = service_image["ref"]
            artifact_service_image["hash"] = service_image["hash"]

        # The metadata from each specified image override is used in the artifact
        elif service_name in context.service_images:
            image_override = context.service_images[service_name]
            context.logger.debug(
                f"Overriding {service_name} image ref to {image_override}."
            )
            artifact_service_image["ref"] = image_override
            artifact_service_image["hash"] = context.match_or_find_hash(
                service_name, image_override
            )

        # By default, the image ref from the manifest is used for the artifact
        else:
            artifact_service_image["ref"] = service_spec["image"]
            artifact_service_image["hash"] = context.match_or_find_hash(
                service_name, service_spec["image"]
            )

    # Check that all service file args match the manifest
    for service_name, service_filename in context.service_files.items():
        if service_name not in context.manifest["services"]:
            # Cannot map specified service name to service in the manifest
            raise ManifestContentException(
                f"Service {service_name} specified to map to service file {service_filename} not found in manifest."
            )

    # Check that all service image args match the manifest
    for service_name, image_override in context.service_images.items():
        if service_name not in context.manifest["services"]:
            # Cannot map specified service name to service in the manifest
            raise ManifestContentException(
                f"Service {service_name} specified to map to service image override {image_override} not found in manifest."
            )

    return artifact_services

prep_artifact_dir

prep_artifact_dir(artifact_prep_dir: Path, images_archive_filename: str, manifests_archive_filename: str, metadata_filename: str) -> None

Prepare artifact directory with necessary files for artifact generation, such as extracted service file images and any other necessary files.

Parameters:

• artifact_prep_dir

  (Path) –

  The directory into which to prepare the artifact files.

• images_archive_filename

  (str) –

  The filename for the artifact images archive.

• manifests_archive_filename

  (str) –

  The filename for the artifact manifests archive.

• metadata_filename

  (str) –

  The filename for the artifact metadata file.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/artifact.py

def prep_artifact_dir(
    self,
    artifact_prep_dir: Path,
    images_archive_filename: str,
    manifests_archive_filename: str,
    metadata_filename: str,
) -> None:
    """
    Prepare artifact directory with necessary files for artifact generation, such as extracted service file images and any other necessary files.

    :param artifact_prep_dir: The directory into which to prepare the artifact files.
    :param images_archive_filename: The filename for the artifact images archive.
    :param manifests_archive_filename: The filename for the artifact manifests archive.
    :param metadata_filename: The filename for the artifact metadata file.
    :return: None
    """

    self.prep_manifests(artifact_prep_dir, manifests_archive_filename)
    self.prep_images(artifact_prep_dir, images_archive_filename)
    metadata_file = artifact_prep_dir / metadata_filename
    with open(metadata_file, "w") as f:
        json.dump(
            {
                "application_name": self.context.manifest_name,
                "orchestrator": "docker-compose",
                "platform": self.context.platform,
                "version": self.artifact_metadata.version,
                "images": self.image_ids,
            },
            f,
        )

prep_delta_image

prep_delta_image(service_name: str, service_image: dict[str, str], artifact_image_prep_dir: Path) -> None

Prepare a delta image file, which may be an empty delta file if the image ref and hash for the service match those in the previous artifact metadata, or a real delta.

Parameters:

• service_name

  (str) –

  The name of the service in the manifest for which to generate an image delta.

• service_image

  (dict[str, str]) –

  The metadata of the image for which to generate a delta.

• artifact_image_prep_dir

  (Path) –

  The directory into which the image delta file will be written.

Returns:

• None –

  None

Raises:

• ImageDeltaException –

  If the delta generation fails.

Source code in src/mender_docker_lifecycle_helper/artifact.py

def prep_delta_image(
    self,
    service_name: str,
    service_image: dict[str, str],
    artifact_image_prep_dir: Path,
) -> None:
    """
    Prepare a delta image file, which may be an empty delta file if the image ref and hash for the service match those in the previous artifact metadata, or a real delta.

    :param service_name: The name of the service in the manifest for which to generate an image delta.
    :param service_image: The metadata of the image for which to generate a delta.
    :param artifact_image_prep_dir: The directory into which the image delta file will be written.
    :raises ImageDeltaException: If the delta generation fails.
    :return: None
    """
    self.context.logger.debug(f"Preparing image delta for service {service_name}")
    # Generate a delta file in the artifact. If the image ref and hash for a service match those in the previous artifact metadata, each layer diff will be empty.
    previous_image = self.context.previous_artifact_metadata.services[service_name][
        "image"
    ]
    (artifact_image_prep_dir / DEEP_DELTA_FILENAME).touch()
    # Ensure images for delta are extracted in the cache
    image_delta_file = self.context.image_cache.delta(previous_image, service_image)
    (artifact_image_prep_dir / IMAGE_FILE_NAME).symlink_to(image_delta_file)
    (artifact_image_prep_dir / "sums-current.txt").write_text(
        previous_image["hash"]
    )
    (artifact_image_prep_dir / "url-current.txt").write_text(previous_image["ref"])

prep_image

prep_image(service_name: str, service_image: dict[str, str], artifact_images_prep_dir: Path) -> None

Prepare the required directory and files for the specified image in the specified artifact prep directory.

Parameters:

• service_name

  (str) –

  The name of the service in the manifest for which to generate an image delta.

• service_image

  (dict[str, str]) –

  The metadata of the image for which to generate a delta.

• artifact_images_prep_dir

  (Path) –

  The directory into which the image directory will be created.

Returns:

• None –

  None

Raises:

• ImageDeltaException –

  If delta generation fails.

Source code in src/mender_docker_lifecycle_helper/artifact.py

def prep_image(
    self,
    service_name: str,
    service_image: dict[str, str],
    artifact_images_prep_dir: Path,
) -> None:
    """
    Prepare the required directory and files for the specified image in the specified artifact prep directory.

    :param service_name: The name of the service in the manifest for which to generate an image delta.
    :param service_image: The metadata of the image for which to generate a delta.
    :param artifact_images_prep_dir: The directory into which the image directory will be created.
    :raises ImageDeltaException: If delta generation fails.
    :return: None
    """
    artifact_image_prep_dir = artifact_images_prep_dir / service_image["hash"]
    self.context.logger.debug(
        f"Preparing artifact files for image {service_image['ref']} for service {service_name} in {artifact_image_prep_dir}"
    )
    # If another service has already prepared this image, just ensure the metadata also matches this service
    if artifact_image_prep_dir.is_dir():
        # If metadata does not match, error (the client will not unpack the same image to multiple refs)
        prepared_sum_new = (artifact_image_prep_dir / "sums-new.txt").read_text()
        prepared_url_new = (artifact_image_prep_dir / "url-new.txt").read_text()
        if (
            # The sum check is effectively redundant, since it always matches the dirname
            prepared_sum_new != service_image["hash"]
            or prepared_url_new != service_image["ref"]
        ):
            raise ManifestContentException(
                f"The image {prepared_sum_new} could not be prepared for {service_image['ref']} for service {service_name} as it is already assigned to {prepared_url_new} and the client will not unpack the same image hash for multiple refs."
            )
        if self.context.delta:
            # If this is a delta image and the prepared image dir is not, error
            if not (artifact_image_prep_dir / DEEP_DELTA_FILENAME).is_file():
                raise ManifestContentException(
                    f"The image {prepared_sum_new} could not be prepared for service {service_name} as a delta as it is already assigned to {prepared_url_new} as non-delta."
                )

            # If this is a delta image and the prepared previous hash is not the same as the previous hash for this, error
            prepared_sum_current = (
                artifact_image_prep_dir / "sums-current.txt"
            ).read_text()
            new_sum_current = self.context.previous_artifact_metadata.services[
                service_name
            ]["image"]["hash"]
            if prepared_sum_current != new_sum_current:
                raise ManifestContentException(
                    f"The image {prepared_sum_new} could not be prepared for service {service_name} as a delta from {new_sum_current} as it already exists as a delta from {prepared_sum_current}"
                )
        # The prepared image metadata matches that which is needed for this image; skip further prep
        return None
    else:
        artifact_image_prep_dir.mkdir()
        (artifact_image_prep_dir / "sums-new.txt").write_text(service_image["hash"])
        (artifact_image_prep_dir / "url-new.txt").write_text(service_image["ref"])

    # For non-delta artifacts or new services, the image as a whole is included in the artifact
    if (
        self.context.delta
        and service_name in self.context.previous_artifact_metadata.services
    ):
        try:
            self.prep_delta_image(
                service_name, service_image, artifact_image_prep_dir
            )
            return
        except ImageDeltaException as e:
            # TODO ensure possible to mix delta and non-delta images in artifact
            self.context.logger.warning("Delta image generation failure:")
            self.context.logger.warning(e)
            self.context.logger.warning(
                f"Will include the full image {service_image} for {service_name} instead."
            )

    # Write the new image details as current in the case of a non-delta image.
    (artifact_image_prep_dir / "sums-current.txt").write_text(service_image["hash"])
    (artifact_image_prep_dir / "url-current.txt").write_text(service_image["ref"])
    self.context.logger.debug(
        f"Including full image {service_image['ref']} with hash {service_image['hash']} for service {service_name} in artifact."
    )
    (artifact_image_prep_dir / IMAGE_FILE_NAME).symlink_to(
        self.context.image_cache.save_cache_image(service_image)
    )

prep_images

prep_images(artifact_prep_dir: Path, images_archive_filename: str) -> None

Prepare the archive file for the specified images in the Mender artifact format, downloading and/or diffing images as necessary based on the image refs and hashes.

Parameters:

• artifact_prep_dir

  (Path) –

  The directory into which the images archive will be generated.

• images_archive_filename

  (str) –

  The name of the images archive file to generate.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/artifact.py

def prep_images(
    self,
    artifact_prep_dir: Path,
    images_archive_filename: str,
) -> None:
    """
    Prepare the archive file for the specified images in the Mender artifact format, downloading and/or diffing images as necessary based on the image refs and hashes.

    :param artifact_prep_dir: The directory into which the images archive will be generated.
    :param images_archive_filename: The name of the images archive file to generate.
    :return: None
    """
    artifact_images_prep_dir = artifact_prep_dir / "images_prep"
    self.context.logger.debug(
        f"Preparing images artifact files in {artifact_images_prep_dir}"
    )

    artifact_images_prep_dir.mkdir()
    for service_name, service_spec in self.artifact_metadata.services.items():
        self.prep_image(
            service_name, service_spec["image"], artifact_images_prep_dir
        )

    images_archive = artifact_prep_dir / images_archive_filename
    self.context.logger.debug(f"Archiving images to file {images_archive}")
    with tarfile.open(
        name=images_archive, mode="w:gz", dereference=True
    ) as images_tar:
        images_tar.add(
            artifact_images_prep_dir,
            arcname="images",
        )
    self.context.logger.debug(
        f"Cleaning up images prep dir {artifact_images_prep_dir}"
    )
    shutil.rmtree(artifact_images_prep_dir)

prep_manifests

prep_manifests(artifact_prep_dir: Path, manifests_archive_filename: str) -> None

Prepare the manifests archive file in the Mender artifact format.

Parameters:

• artifact_prep_dir

  (Path) –

  The directory into which the manifests archive will be generated.

• manifests_archive_filename

  (str) –

  The name of the manifests archive file to generate.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/artifact.py

def prep_manifests(
    self, artifact_prep_dir: Path, manifests_archive_filename: str
) -> None:
    """
    Prepare the manifests archive file in the Mender artifact format.

    :param artifact_prep_dir: The directory into which the manifests archive will be generated.
    :param manifests_archive_filename: The name of the manifests archive file to generate.
    :return: None
    """
    artifact_manifests_prep_dir = artifact_prep_dir / "manifests_prep"
    self.context.logger.debug(
        f"Preparing manifests artifact files in {artifact_manifests_prep_dir}"
    )

    artifact_manifests_prep_dir.mkdir()
    artifact_manifest = (
        artifact_manifests_prep_dir / self.context.manifest_file.name
    )
    with open(artifact_manifest, "w") as f:
        yaml.safe_dump(self.context.manifest, f)

    manifests_archive = artifact_prep_dir / manifests_archive_filename
    with tarfile.open(name=manifests_archive, mode="w:gz") as manifests_tar:
        manifests_tar.add(
            artifact_manifests_prep_dir,
            arcname="manifests",
        )
    self.context.logger.debug(
        f"Cleaning up manifests prep dir {artifact_manifests_prep_dir}"
    )
    shutil.rmtree(artifact_manifests_prep_dir)

mender_docker_lifecycle_helper.artifact_metadata

Classes:

• ArtifactMetadata –

ArtifactMetadata

ArtifactMetadata(version: str, services: dict[str, dict[str, dict[str, str]]] | None)

Construct an ArtifactMetadata object.

Parameters:

• version

  (str) –

  The version identifier for the artifact.

• services

  (dict[str, dict[str, dict[str, str]]] | None) –

  The metadata of the services included in the artifact. The structure of this metadata is: { serviceName: { image: { ref: str, hash: str } } }

Returned by:

• API reference  context  LifecycleHelperContext  _prep_previous_artifact_metadata

Used by:

• API reference
  •  artifact  LifecycleHelperArtifact
  •  helper  LifecycleHelper
    •  create_artifact
    •  update_cache_artifact_metadata

Methods:

• from_dict –

  Construct an ArtifactMetadata object directly from a dict.

• from_file –

  Construct an ArtifactMetadata object directly from a file.

• to_dict –

  Dump the contents of the artifact metadata as a dict.

• to_file –

  Dump the contents of the artifact metadata to a file, as JSON structured as the to_dict return value.

Source code in src/mender_docker_lifecycle_helper/artifact_metadata.py

def __init__(
    self, version: str, services: dict[str, dict[str, dict[str, str]]] | None
):
    """
    Construct an ArtifactMetadata object.

    :param version: The version identifier for the artifact.
    :param services: The metadata of the services included in the artifact. The structure of this metadata is:
        {
            serviceName: {
                image: {
                    ref: str,
                    hash: str
                }
            }
        }
    """
    self.version = version
    self.services = services if services is not None else {}

from_dict classmethod

from_dict(data: dict[str, dict[str, dict[str, dict[str, str]]]])

Construct an ArtifactMetadata object directly from a dict.

Parameters:

• data

  (dict[str, dict[str, dict[str, dict[str, str]]]]) –

  The metadata for the artifact, structured as the to_dict return value.

Returns:

• –

  An object constructed from the provided data.

Source code in src/mender_docker_lifecycle_helper/artifact_metadata.py

@classmethod
def from_dict(cls, data: dict[str, dict[str, dict[str, dict[str, str]]]]):
    """
    Construct an ArtifactMetadata object directly from a dict.

    :param data: The metadata for the artifact, structured as the to_dict return value.
    :return: An object constructed from the provided data.
    """
    return cls(version=data.get("version"), services=data.get("services", {}))

from_file classmethod

from_file(file_path: Path)

Construct an ArtifactMetadata object directly from a file.

Parameters:

• file_path

  (Path) –

  The path to a JSON file containing the artifact metadata, structured as the to_dict return value.

Returns:

• –

  An object constructed from the provided file.

Source code in src/mender_docker_lifecycle_helper/artifact_metadata.py

@classmethod
def from_file(cls, file_path: Path):
    """
    Construct an ArtifactMetadata object directly from a file.

    :param file_path: The path to a JSON file containing the artifact metadata, structured as the to_dict return value.
    :return: An object constructed from the provided file.
    """
    with open(file_path, "r") as f:
        data = json.load(f)
    return cls.from_dict(data)

to_dict

to_dict()

Dump the contents of the artifact metadata as a dict.

Returns:

• –

  The metadata of the artifact, structured as: { version: str, services: { ... (see services param in init) } }

Source code in src/mender_docker_lifecycle_helper/artifact_metadata.py

def to_dict(self):
    """
    Dump the contents of the artifact metadata as a dict.

    :return: The metadata of the artifact, structured as:
        {
            version: str,
            services: {
                ... (see services param in __init__)
            }
        }
    """
    return {"version": self.version, "services": self.services}

to_file

to_file(file_path: Path)

Dump the contents of the artifact metadata to a file, as JSON structured as the to_dict return value.

Parameters:

• file_path

  (Path) –

  The path to which to dump the metadata of the artifact.

Returns:

• –

  None

Source code in src/mender_docker_lifecycle_helper/artifact_metadata.py

def to_file(self, file_path: Path):
    """
    Dump the contents of the artifact metadata to a file, as JSON structured as the to_dict return value.

    :param file_path: The path to which to dump the metadata of the artifact.
    :return: None
    """
    file_path.parent.mkdir(parents=True, exist_ok=True)
    with open(file_path, "w") as f:
        json.dump(self.to_dict(), f, indent=2)

mender_docker_lifecycle_helper.cli

Functions:

• cli –

  Produce and deploy a Mender artifact for the MANIFEST_FILE (compose yaml) Docker application, as deltas against local cache when available or repo version context otherwise.

cli

cli(**args) -> None

Produce and deploy a Mender artifact for the MANIFEST_FILE (compose yaml) Docker application, as deltas against local cache when available or repo version context otherwise.

Source code in src/mender_docker_lifecycle_helper/cli.py

@click.command(context_settings={"show_default": True})
@click.version_option()
@click.option(
    "--clean-cache",
    is_flag=True,
    help="ONLY clean the cache directory (based on --cache-limit-size or --cache-limit-percent); do not perform any artifact operations.",
)
@click.option(
    "--clear-image-cache",
    is_flag=True,
    help="Remove all image cache contents (save, extract, delta); do not perform any artifact operations.",
)
@click.option(
    "--clear-cache",
    is_flag=True,
    help="Remove the entire cache directory; do not perform any artifact operations.",
)
@click.option(
    "--cache-limit-size",
    default=None,
    type=int,
    help="Maximum cache size in bytes. When exceeded, oldest items are removed.",
)
@click.option(
    "--cache-limit-percent",
    default=20.0,
    type=float,
    help="Minimum percent of total disk that should remain free.",
)
@click.option(
    "-a",
    "--artifact-filename",
    default=None,
    help="Name of the artifact file to create. [default: <manifest-name>-<previous-version>+<current repo commit SHA>+<UUID>.mender]",
    type=str,
    show_default=False,
)
@click.option(
    "--cache/--no-cache",
    default=True,
    flag_value=True,
    help="Read/skip reading previous artifact info from cache and always read from the repo at the previous version.",
)
@click.option(
    "--cache-dir",
    default=LifecycleHelperContext._default_cache_dir(),
    help="The cache dir for the helper. Overrides the MENDER_HELPER_CACHE_DIR variable.",
    type=click.Path(file_okay=False, dir_okay=True),
)
@click.option(
    "--cache-limit/--no-cache-limit",
    default=True,
    help="Enable automatic cache cleanup when size limits are exceeded.",
)
@click.option(
    "--delta/--no-delta",
    default=True,
    flag_value=True,
    help="Generate the artifact as an update artifact, if applicable.",
)
@click.option(
    "-t",
    "--device-type",
    help="Device type for the artifact. [required]",
    type=str,
)
@click.option(
    "-g",
    "--device-group",
    default=None,
    help="Device group to which to deploy the artifact, or skip deployment if not defined.",
    type=str,
)
@click.option(
    "-l",
    "--log-level",
    default="INFO",
    help="Set logging level",
    type=click.Choice(LOG_LEVELS, case_sensitive=False),
)
@click.option(
    "-m",
    "--manifest-name",
    default=None,
    help="The application/software name for the artifact [default: <dirname of repo containing manifest_file>-<dirname directly containing manifest_file>].",
    show_default=False,
    type=str,
)
@click.option(
    "-h",
    "--mender-host",
    default="https://hosted.mender.io",
    help="Mender host URL for artifact upload and deployment.",
)
@click.option(
    "-p",
    "--platform",
    help="Platform with which the artifact is compatible (e.g., linux/arm/v7) [required]",
    type=str,
)
@click.option(
    "--previous-version",
    default=None,
    help="Repo ref from which to read image names and versions for comparison to the current state. [default: contents of the VERSION file in root of the repo containing manifest_file]",
    show_default=False,
    type=str,
)
@click.option(
    "-r",
    "--release/--no-release",
    default=False,
    flag_value=True,
    help="Create the artifact for a release, using the current value of the VERSION file as the artifact version and the value of the VERSION file at the previous commit as the --previous-version.",
)
@click.option(
    "-f",
    "--service-file",
    "service_files",
    default=None,
    metavar="<SERVICE NAME> <IMAGE FILE>",
    help="Image file to extract and use to override the image for the specified service in the manifest_file. Can be specified multiple times.",
    type=click.Tuple([str, str]),
    multiple=True,
)
@click.option(
    "-i",
    "--service-image",
    "service_images",
    default=None,
    metavar="<SERVICE NAME> <IMAGE NAME>",
    help="Image name to override for the specified service in the manifest_file. Can be specified multiple times.",
    type=click.Tuple([str, str]),
    multiple=True,
)
@click.option(
    "-w",
    "--wait-for-deploy/--no-wait-for-deploy",
    is_flag=True,
    default=True,
    help="Wait for deployment to complete and only cache metadata on success.",
)
@click.option(
    "-v",
    "--verbose",
    count=True,
    help="Increase verbosity by one level (see --log-level). Can be specified multiple times.",
)
@click.argument(
    "manifest_file",
    type=click.Path(
        exists=True,
        file_okay=True,
        dir_okay=False,
        readable=True,
        allow_dash=True,
        path_type=Path,
    ),
    required=False,
)
def cli(**args) -> None:
    """
    Produce and deploy a Mender artifact for the MANIFEST_FILE (compose yaml) Docker application, as deltas against local cache when available or repo version context otherwise.
    """

    args = SimpleNamespace(**args)

    # Cache limit size takes precedence over percent
    args.cache_limit_percent = (
        None if args.cache_limit_size is not None else args.cache_limit_percent
    )

    args.cache_operation_only = (
        args.clear_cache or args.clear_image_cache or args.clean_cache
    )
    args.log_level = LOG_LEVELS[max(0, LOG_LEVELS.index(args.log_level) - args.verbose)]

    service_files = {}
    for service, file in args.service_files:
        service_files[service] = file
    args.service_files = service_files

    service_images = {}
    for service, image in args.service_images:
        service_images[service] = image
    args.service_images = service_images

    helper = LifecycleHelper(args)

    if args.cache_operation_only:
        return

    if not args.manifest_file:
        raise click.ClickException("manifest_file is required")
    if not args.platform:
        raise click.ClickException("--platform is required")
    if not args.device_type:
        raise click.ClickException("--device-type is required")

    helper.prep_artifact()

mender_docker_lifecycle_helper.context

Classes:

• LifecycleHelperContext –

  Representation of the context in which the helper is operating, including the args, input file, and available container images.

LifecycleHelperContext

LifecycleHelperContext(args: SimpleNamespace)

Representation of the context in which the helper is operating, including the args, input file, and available container images.

Construct a LifecycleHelperContext object.

Parameters:

• args

  (SimpleNamespace) –

  An object of the args for the helper execution. See cli.py.

Used by:

• API reference
  •  artifact  LifecycleHelperArtifact
  •  mender_server
    •  call_mender_host_api
    •  get_deployment_status
    •  upload_artifact
    •  wait_for_deployment

Methods:

• clean_cache –

  Clean the image cache based on size or disk percent limits.

• clear_cache –

  Remove the entire cache directory.

• clear_image_cache –

  Remove the image cache contents (save, extract, delta).

• match_or_find_hash –

  If the provided image ref matches that in the previous artifact metadata for the given service, return the corresponding hash from the previous artifact metadata. Otherwise, attempt to retrieve the image hash for the provided image ref.

Source code in src/mender_docker_lifecycle_helper/context.py

def __init__(self, args: SimpleNamespace):
    """
    Construct a LifecycleHelperContext object.

    :param args: An object of the args for the helper execution. See cli.py.
    """
    self.artifact_filename = args.artifact_filename
    self.cache = args.cache
    self.cache_limit = args.cache_limit
    self.cache_limit_percent = args.cache_limit_percent
    self.cache_limit_size = args.cache_limit_size
    self.delta = args.delta
    self.device_type = args.device_type
    self.device_group = args.device_group
    self.mender_host = args.mender_host
    self.mender_pat = os.getenv("MENDER_PAT", None)
    self.platform = args.platform
    self.release = args.release
    self.service_files = args.service_files
    self.service_images = args.service_images
    self.wait_for_deploy = args.wait_for_deploy

    self.logger = self._prep_logger(args.log_level)

    if args.cache_operation_only:
        self.cache_dir = self._prep_cache_dir(args.cache_dir)
        self.image_cache = ImageCache(
            self.cache_dir / "images", platform=self.platform, logger=self.logger
        )

        if args.clear_cache:
            self.clear_cache()

        if args.clear_image_cache:
            self.clear_image_cache()

        if args.clean_cache:
            if args.cache_limit_size is None and args.cache_limit_percent is None:
                self.logger.error(
                    "No cleanup limit specified. Use --cache-limit-size or --cache-limit-percent."
                )
                return

            bytes_freed = self.clean_cache(
                args.cache_limit_size, args.cache_limit_percent
            )
            freed = float(bytes_freed)
            base = 1000
            units = ["B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"]
            idx = 0
            while freed >= base and idx < len(units) - 1:
                freed /= base
                idx += 1
            self.logger.info(f"Cache cleanup freed {freed:.{2}f} {units[idx]}.")
        return

    self.manifest_file = args.manifest_file.resolve()

    self.repo_root_dir = self._repo_root_dir(self.manifest_file)
    self.repo_version = self._repo_version(self.repo_root_dir)
    self._repo = git.Repo(self.repo_root_dir)
    self.commit_short_sha = self._repo.head.commit.hexsha[:7]

    self.manifest_name = (
        args.manifest_name
        if args.manifest_name
        else f"{self.repo_root_dir.name}-{self.manifest_file.parent.name}"
    )

    if self.cache:
        self.cache_dir = self._prep_cache_dir(args.cache_dir)
        manifests_cache_dir = self.cache_dir / "manifests"
        manifests_cache_dir.mkdir(exist_ok=True)
        manifest_cache_dir = manifests_cache_dir / self.manifest_name
        manifest_cache_dir.mkdir(exist_ok=True)
        self.cache_artifact_metadata_file = (
            manifest_cache_dir / "previous_artifact.json"
        )
    else:
        self.cache_dir = Path(
            self.repo_root_dir / ".mender-docker-lifecycle-helper"
        )
    self.image_cache = ImageCache(
        self.cache_dir / "images", platform=self.platform, logger=self.logger
    )
    self.temp_dir = self.cache_dir / "temp"
    self.temp_dir.mkdir(parents=True)

    self.manifest = self._compose_content_from_file(self.manifest_file)

    if self.delta:
        self.previous_artifact_metadata = self._prep_previous_artifact_metadata(
            args.previous_version
        )

_artifact_services_metadata_from_compose

_artifact_services_metadata_from_compose(compose_file: Path) -> dict[str, dict[str, dict[str, str]]]

Extract the services metadata from an artifact compose manifest file.

Parameters:

• compose_file

  (Path) –

  The path to the compose manifest file from which to extract metadata.

Returns:

• dict[str, dict[str, dict[str, str]]] –

  The metadata of the services expressed in the compose manifest file, in the following format: { serviceName: { image: { ref: str, hash: str } } }

Source code in src/mender_docker_lifecycle_helper/context.py

def _artifact_services_metadata_from_compose(
    self, compose_file: Path
) -> dict[str, dict[str, dict[str, str]]]:
    """
    Extract the services metadata from an artifact compose manifest file.

    :param compose_file: The path to the compose manifest file from which to extract metadata.
    :return: The metadata of the services expressed in the compose manifest file, in the following format:
        {
            serviceName: {
                image: {
                    ref: str,
                    hash: str
                }
            }
        }
    """
    compose = self._compose_content_from_file(compose_file)

    return {
        service: {
            "image": {
                "ref": config["image"],
                "hash": get_image_hash(config["image"], self.logger),
            }
        }
        for service, config in compose["services"].items()
    }

_compose_content_from_file

_compose_content_from_file(compose_file: Path) -> dict

Read the contents of a Compose YAML file using all expansions per the spec.

Parameters:

• compose_file

  (Path) –

  The path to the Compose YAML file to read.

Returns:

• dict –

  The normalized compose content as a dict with all directives resolved.

Raises:

• RuntimeError –

  If docker compose config fails.

• yaml.YAMLError –

  If the output cannot be parsed as YAML.

Source code in src/mender_docker_lifecycle_helper/context.py

def _compose_content_from_file(self, compose_file: Path) -> dict:
    """
    Read the contents of a Compose YAML file using all expansions per the spec.

    :param compose_file: The path to the Compose YAML file to read.
    :return: The normalized compose content as a dict with all directives resolved.
    :raises RuntimeError: If docker compose config fails.
    :raises yaml.YAMLError: If the output cannot be parsed as YAML.
    """
    try:
        result = subprocess.run(
            [
                DOCKER_BIN,
                "compose",
                "--file",
                str(compose_file),
                "config",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
    except subprocess.CalledProcessError as e:
        error_msg = f"Failed to read compose file {compose_file}: {e.stderr}"
        raise RuntimeError(error_msg) from e
    return yaml.safe_load(result.stdout)

_default_cache_dir staticmethod

_default_cache_dir(cache_dir_env_key: str = 'MENDER_HELPER_CACHE_DIR', default_cache_dir_name: str = 'mender-docker-lifecycle-helper') -> Path

Determine the path use for the default helper cache dir based on the values of relevant env vars.

Parameters:

• cache_dir_env_key

  (str, default: 'MENDER_HELPER_CACHE_DIR' ) –

  The env var from which to read the user-specified helper cache dir, if defined, defaults to "MENDER_HELPER_CACHE_DIR"

• default_cache_dir_name

  (str, default: 'mender-docker-lifecycle-helper' ) –

  The dir name of the helper cache within general cache dir, defaults to "mender-docker-lifecycle-helper"

Returns:

• Path –

  The path to the default helper cache dir.

Source code in src/mender_docker_lifecycle_helper/context.py

@staticmethod
def _default_cache_dir(
    cache_dir_env_key: str = "MENDER_HELPER_CACHE_DIR",
    default_cache_dir_name: str = "mender-docker-lifecycle-helper",
) -> Path:
    """
    Determine the path use for the default helper cache dir based on the values of relevant env vars.

    :param cache_dir_env_key: The env var from which to read the user-specified helper cache dir, if defined, defaults to "MENDER_HELPER_CACHE_DIR"
    :param default_cache_dir_name: The dir name of the helper cache within general cache dir, defaults to "mender-docker-lifecycle-helper"
    :return: The path to the default helper cache dir.
    """
    return (
        Path(os.getenv(cache_dir_env_key))
        if os.getenv(cache_dir_env_key)
        else (
            Path(os.getenv("XDG_CACHE_HOME")) / default_cache_dir_name
            if os.getenv("XDG_CACHE_HOME")
            else Path("~/.cache").expanduser() / default_cache_dir_name
        )
    )

_prep_cache_dir

_prep_cache_dir(cache_dir: Path) -> Path

Prepare the cache directory for the helper execution, creating if necessary.

Parameters:

• cache_dir

  (Path) –

  The directory of the cache dir to use or create.

Returns:

• Path –

  The path to the created or already existing cache dir.

Source code in src/mender_docker_lifecycle_helper/context.py

def _prep_cache_dir(self, cache_dir: Path) -> Path:
    """
    Prepare the cache directory for the helper execution, creating if necessary.

    :param cache_dir: The directory of the cache dir to use or create.
    :return: The path to the created or already existing cache dir.
    """
    if cache_dir.exists():
        self.logger.debug(f"Using existing cache dir {cache_dir}")
    else:
        self.logger.warning(f"Cache dir {cache_dir} does not exist, will create...")
        cache_dir.mkdir(parents=True)
    return cache_dir

_prep_logger staticmethod

_prep_logger(log_level: str) -> Logger

Prepare a logger object for the helper execution at the specified level.

Parameters:

• log_level

  (str) –

  The log level for the logger; must be a string matching a logging level attribute, e.g. "INFO".

Returns:

• Logger –

  A logger object.

Source code in src/mender_docker_lifecycle_helper/context.py

@staticmethod
def _prep_logger(log_level: str) -> logging.Logger:
    """
    Prepare a logger object for the helper execution at the specified level.

    :param log_level: The log level for the logger; must be a string matching a logging level attribute, e.g. "INFO".
    :return: A logger object.
    """
    logger = logging.getLogger(__name__)
    handler = logging.StreamHandler()
    handler.setFormatter(
        ColorLogFormatter("[%(asctime)s] %(levelname)s: %(message)s")
    )
    logger.addHandler(handler)
    logger.setLevel(getattr(logging, log_level))
    return logger

_prep_previous_artifact_metadata

_prep_previous_artifact_metadata(previous_version: str | None) -> ArtifactMetadata

Determine the metadata of the previous artifact. If the cache is enabled and includes a metadata file from a previous helper execution, that data is used. If a previous version is specified, the metadata is extracted from the artifact compose manifest file at that version of the repository. If the execution is for a release, the metadata is extracted from the artifact compose manifest file at the previous (mainline) commit of the repo. Otherwise, the metadata is extracted from the artifact compose manifest file at the version of the repository as specified by the current repo version (as read from the version file).

Parameters:

• previous_version

  (str | None) –

  If provided, the version of the helper execution repository from which to read the previous artifact metadata.

Returns:

• ArtifactMetadata –

  The metadata of the previous artifact, in the format as returned by _artifact_services_metadata_from_compose.

Raises:

• FileNotFoundError –

  If the manifest file does not exist.

Source code in src/mender_docker_lifecycle_helper/context.py

def _prep_previous_artifact_metadata(
    self,
    previous_version: str | None,
) -> ArtifactMetadata:
    """
    Determine the metadata of the previous artifact. If the cache is enabled and includes a metadata file from a previous helper execution, that data is used. If a previous version is specified, the metadata is extracted from the artifact compose manifest file at that version of the repository. If the execution is for a release, the metadata is extracted from the artifact compose manifest file at the previous (mainline) commit of the repo. Otherwise, the metadata is extracted from the artifact compose manifest file at the version of the repository as specified by the current repo version (as read from the version file).

    :param previous_version: If provided, the version of the helper execution repository from which to read the previous artifact metadata.
    :raises FileNotFoundError: If the manifest file does not exist.
    :return: The metadata of the previous artifact, in the format as returned by _artifact_services_metadata_from_compose.
    """
    if self.cache and self.cache_artifact_metadata_file.exists():
        self.logger.debug(
            f"Cached artifact metadata file {self.cache_artifact_metadata_file} found, will use this for previous artifact metadata."
        )
        return ArtifactMetadata.from_file(self.cache_artifact_metadata_file)

    if previous_version:
        # If a version/ref is provided, attempt to get the artifact info from that ref
        self.logger.info(
            f"Getting previous artifact info for arg-specified repo ref: {previous_version}"
        )
    elif self.release:
        previous_commit = self._repo.head.commit.parents[0].hexsha
        self.logger.info(
            f"Release context indicated, will read the previous version from the repo at ref {previous_commit}"
        )
        previous_version = self._repo_version(
            self._temp_repo_at_version(previous_commit)
        )
    else:
        previous_version = self.repo_version
        self.logger.debug(
            f"Getting previous artifact info from repo at version from VERSION file: {previous_version}"
        )

    previous_version_repo = self._temp_repo_at_version(previous_version)
    previous_manifest_file = previous_version_repo / self.manifest_file.relative_to(
        self.repo_root_dir
    )
    if not previous_manifest_file.exists():
        self.logger.error(f"Manifest file {previous_manifest_file} does not exist.")
        raise FileNotFoundError(
            f"Manifest file {previous_manifest_file} does not exist."
        )

    return ArtifactMetadata(
        version=previous_version,
        services=self._artifact_services_metadata_from_compose(
            previous_manifest_file
        ),
    )

_repo_root_dir staticmethod

_repo_root_dir(path: Path) -> Path

Find the root directory of the Git repo containing the specified path.

Parameters:

• path

  (Path) –

  The path for which to find the containing repo.

Returns:

• Path –

  The path to the containing Git repo.

Raises:

• FileNotFoundError –

  If the repository root directory cannot be found.

Source code in src/mender_docker_lifecycle_helper/context.py

@staticmethod
def _repo_root_dir(path: Path) -> Path:
    """
    Find the root directory of the Git repo containing the specified path.

    :param path: The path for which to find the containing repo.
    :raises FileNotFoundError: If the repository root directory cannot be found.
    :return: The path to the containing Git repo.
    """
    while path != Path("/"):
        if (path / ".git").exists():
            return path
        path = path.parent
    raise FileNotFoundError("Could not find the repository root directory.")

_repo_version staticmethod

_repo_version(repo_dir: Path) -> str

Find the version of the specified repo per its version file.

Parameters:

• repo_dir

  (Path) –

  The path to the repository directory in which to find the version file.

Returns:

• str –

  The version as specified in the version file.

Source code in src/mender_docker_lifecycle_helper/context.py

@staticmethod
def _repo_version(repo_dir: Path) -> str:
    """
    Find the version of the specified repo per its version file.

    :param repo_dir: The path to the repository directory in which to find the version file.
    :return: The version as specified in the version file.
    """
    VERSION_FILE_NAME = "VERSION"
    with open(repo_dir / VERSION_FILE_NAME, "r") as file:
        return file.read().strip()

_temp_repo_at_version

_temp_repo_at_version(version: str) -> Path

Prepare a temporary clone of the helper execution repo at the specified version.

Parameters:

• version

  (str) –

  The version at which to clone the repo.

Returns:

• Path –

  The path to the temporary clone of the repo.

Raises:

• Exception –

  If the repository clone or checkout fails.

Source code in src/mender_docker_lifecycle_helper/context.py

def _temp_repo_at_version(
    self,
    version: str,
) -> Path:
    """
    Prepare a temporary clone of the helper execution repo at the specified version.

    :param version: The version at which to clone the repo.
    :raises Exception: If the repository clone or checkout fails.
    :return: The path to the temporary clone of the repo.
    """
    temp_repo_dir = Path(
        tempfile.mkdtemp(dir=(self.temp_dir if hasattr(self, "temp_dir") else None))
    )
    self.logger.debug(
        f"Preparing temporary repo at version {version}: {temp_repo_dir}"
    )

    # Clone the repo from the local path to the temporary directory and checkout the specified version
    try:
        repo = git.Repo.clone_from(self.repo_root_dir, temp_repo_dir)
        repo.git.checkout(version)
        self.logger.debug(
            f"Cloned and checked out repo at {version} to {temp_repo_dir}"
        )
    except Exception as e:
        self.logger.error(
            f"Failed to clone and checkout repo at {version} in {temp_repo_dir}: {e}"
        )
        raise e

    return temp_repo_dir

clean_cache

clean_cache(limit_size_bytes: int | None, disk_percent: float | None) -> int

Clean the image cache based on size or disk percent limits.

Parameters:

• limit_size_bytes

  (int | None) –

  Maximum cache size in bytes.

• disk_percent

  (float | None) –

  Minimum percent of total disk that should remain free.

Returns:

• int –

  Number of bytes freed.

Source code in src/mender_docker_lifecycle_helper/context.py

def clean_cache(
    self, limit_size_bytes: int | None, disk_percent: float | None
) -> int:
    """
    Clean the image cache based on size or disk percent limits.

    :param limit_size_bytes: Maximum cache size in bytes.
    :param disk_percent: Minimum percent of total disk that should remain free.
    :return: Number of bytes freed.
    """
    return self.image_cache.cleanup_by_mtime(
        limit_size_bytes=limit_size_bytes,
        disk_percent=disk_percent,
    )

clear_cache

clear_cache() -> None

Remove the entire cache directory.

Raises:

• FileNotFoundError –

  If the cache directory does not exist.

Source code in src/mender_docker_lifecycle_helper/context.py

def clear_cache(self) -> None:
    """
    Remove the entire cache directory.

    :raises FileNotFoundError: If the cache directory does not exist.
    """
    if self.cache_dir.exists():
        shutil.rmtree(self.cache_dir)
    else:
        raise FileNotFoundError(f"Cache directory does not exist: {self.cache_dir}")

clear_image_cache

clear_image_cache() -> None

Remove the image cache contents (save, extract, delta).

Raises:

• FileNotFoundError –

  If the image cache does not exist.

Source code in src/mender_docker_lifecycle_helper/context.py

def clear_image_cache(self) -> None:
    """
    Remove the image cache contents (save, extract, delta).

    :raises FileNotFoundError: If the image cache does not exist.
    """
    shutil.rmtree(self.image_cache.cache_dir)

match_or_find_hash

match_or_find_hash(service_name: str, image_ref: str) -> str

If the provided image ref matches that in the previous artifact metadata for the given service, return the corresponding hash from the previous artifact metadata. Otherwise, attempt to retrieve the image hash for the provided image ref.

Parameters:

• service_name

  (str) –

  The name of the service for which to look for a matching image ref.

• image_ref

  (str) –

  The ref of the image for which to look for a match.

Returns:

• str –

  The hash of the image with the provided ref.

Source code in src/mender_docker_lifecycle_helper/context.py

def match_or_find_hash(self, service_name: str, image_ref: str) -> str:
    """
    If the provided image ref matches that in the previous artifact metadata for the given service, return the corresponding hash from the previous artifact metadata. Otherwise, attempt to retrieve the image hash for the provided image ref.

    :param service_name: The name of the service for which to look for a matching image ref.
    :param image_ref: The ref of the image for which to look for a match.
    :return: The hash of the image with the provided ref.
    """

    image_hash = ""
    if (
        self.previous_artifact_metadata.services
        if hasattr(self, "previous_artifact_metadata")
        else {}
    ).get(service_name, {}).get("image", {}).get("ref", "") == image_ref:
        image_hash = self.previous_artifact_metadata.services[service_name][
            "image"
        ]["hash"]
        self.logger.debug(
            f"Image ref for service {service_name} matches previous artifact metadata. Skipping hash lookup and using previous hash {image_hash}."
        )
    else:
        image_hash = get_image_hash(image_ref, self.logger)

    return image_hash

mender_docker_lifecycle_helper.helper

Classes:

• LifecycleHelper –

  Representation of the lifecycle helper execution.

LifecycleHelper

LifecycleHelper(args: SimpleNamespace)

Representation of the lifecycle helper execution.

Construct a LifecycleHelper object.

Parameters:

• args

  (SimpleNamespace) –

  An object of the args for the helper execution. See cli.py.

Methods:

• create_artifact –

  Generate the Mender artifact for the specified metadata.

• deploy_artifact –

  Issue a deployment of a pre-uploaded artifact.

• prep_artifact –

  Prepare the artifact, including creation, upload, and deployment as specified by provided args.

• update_cache_artifact_metadata –

  Write the specified artifact metadata to the cache previous metadata location, if applicable.

Source code in src/mender_docker_lifecycle_helper/helper.py

def __init__(self, args: SimpleNamespace):
    """
    Construct a LifecycleHelper object.

    :param args: An object of the args for the helper execution. See cli.py.
    """
    self.context = LifecycleHelperContext(args)

create_artifact

create_artifact(artifact_metadata: ArtifactMetadata) -> LifecycleHelperArtifact

Generate the Mender artifact for the specified metadata.

Parameters:

• artifact_metadata

  (ArtifactMetadata) –

  The metadata for the artifact to generate.

Returns:

• LifecycleHelperArtifact –

  The object representing the generated artifact.

Source code in src/mender_docker_lifecycle_helper/helper.py

def create_artifact(
    self, artifact_metadata: ArtifactMetadata
) -> LifecycleHelperArtifact:
    """
    Generate the Mender artifact for the specified metadata.

    :param artifact_metadata: The metadata for the artifact to generate.
    :return: The object representing the generated artifact.
    """
    artifact_name = f"{self.context.manifest_name}-{artifact_metadata.version}"
    artifact_filename = Path(
        self.context.artifact_filename
        if self.context.artifact_filename is not None
        else f"{artifact_name}.mender"
    ).resolve()
    artifact = LifecycleHelperArtifact(
        self.context, artifact_name, artifact_metadata, artifact_filename
    )
    self.context.logger.info(f"Generating artifact file {artifact_filename}")
    artifact.gen_artifact_file()
    self.context.logger.info("Artifact file generated successfully.")
    return artifact

deploy_artifact

deploy_artifact(artifact: LifecycleHelperArtifact) -> str

Issue a deployment of a pre-uploaded artifact.

Parameters:

• artifact

  (LifecycleHelperArtifact) –

  The object of the artifact to deploy to the Mender server.

Returns:

• str –

  The deployment ID.

Source code in src/mender_docker_lifecycle_helper/helper.py

def deploy_artifact(self, artifact: LifecycleHelperArtifact) -> str:
    """
    Issue a deployment of a pre-uploaded artifact.

    :param artifact: The object of the artifact to deploy to the Mender server.
    :return: The deployment ID.
    """
    deployment_name = f"{artifact.name}-{self.context.device_group}"

    self.context.logger.debug(
        f"Creating deployment for artifact {artifact.name} to device group {self.context.device_group}"
    )
    response = call_mender_host_api(
        self.context,
        f"deployments/deployments/group/{self.context.device_group}",
        {
            "json": {
                "name": deployment_name,
                "artifact_name": artifact.name,
            }
        },
    )
    deployment_id = None
    if "Location" in response.headers:
        location = response.headers["Location"]
        deployment_id = location.rstrip("/").split("/")[-1]
        self.context.logger.debug(
            f"Got deployment ID from Location header: {deployment_id}"
        )
    else:
        self.context.logger.error(
            f"Could not extract deployment ID from response. "
            f"Body: {response.text!r}, Headers: {dict(response.headers)}"
        )
        raise ValueError(
            "Could not determine deployment ID from Mender API response"
        )
    self.context.logger.info(
        f"Created deployment {deployment_name} (ID: {deployment_id})"
    )
    return deployment_id

prep_artifact

prep_artifact() -> None

Prepare the artifact, including creation, upload, and deployment as specified by provided args.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/helper.py

def prep_artifact(self) -> None:
    """
    Prepare the artifact, including creation, upload, and deployment as specified by provided args.

    :return: None
    """
    artifact_version = None
    if self.context.release:
        self.context.logger.info(
            "Preparing an artifact for release, so using the repo version."
        )
        artifact_version = self.context.repo_version
    else:
        artifact_version = f"{self.context.repo_version}+{self.context.commit_short_sha}+{uuid.uuid4()}"
    self.context.logger.info(
        f"Preparing an artifact with the version {artifact_version}"
    )

    artifact_metadata = ArtifactMetadata(
        artifact_version,
        services=LifecycleHelperArtifact.gen_artifact_services(self.context),
    )
    artifact = self.create_artifact(artifact_metadata)
    upload_artifact(self.context, artifact)

    # Start cache cleanup in parallel if limits are configured
    cleanup_thread = None
    cleanup_bytes_freed = [0]

    if self.context.cache_limit and (
        self.context.cache_limit_size is not None
        or self.context.cache_limit_percent is not None
    ):
        self.context.logger.warning(
            "Cache limit exceeded, starting cache cleanup in background..."
        )

        def run_cleanup():
            bytes_freed = self.context.image_cache.cleanup_by_mtime(
                limit_size_bytes=self.context.cache_limit_size,
                disk_percent=self.context.cache_limit_percent,
            )
            cleanup_bytes_freed[0] = bytes_freed

        cleanup_thread = threading.Thread(target=run_cleanup, daemon=False)
        cleanup_thread.start()

    if self.context.device_group is not None:
        deployment_id = self.deploy_artifact(artifact)

        if self.context.wait_for_deploy:
            self.context.logger.info(
                f"Waiting for deployment {deployment_id} to complete..."
            )
            if wait_for_deployment(self.context, deployment_id):
                self.context.logger.debug(
                    f"Deployment {deployment_id} for {artifact.name} succeeded."
                )
                self.update_cache_artifact_metadata(artifact_metadata)
            else:
                raise ArtifactDeployFailure(
                    f"Deployment {deployment_id} did not succeed; "
                    "cached metadata will not be updated."
                )
        else:
            self.context.logger.debug(f"Artifact {artifact.name} deployed.")
            self.update_cache_artifact_metadata(artifact_metadata)
    else:
        self.context.logger.debug(
            "No device group set; skipping deployment creation."
        )

    # Wait for cleanup thread to complete
    if cleanup_thread is not None:
        cleanup_thread.join()
        self.context.logger.info(
            f"Cache cleanup freed {cleanup_bytes_freed[0]} bytes."
        )

    self.context.logger.info(f"Artifact {artifact.name} successfully processed!")

update_cache_artifact_metadata

update_cache_artifact_metadata(artifact_metadata: ArtifactMetadata) -> None

Write the specified artifact metadata to the cache previous metadata location, if applicable.

Parameters:

• artifact_metadata

  (ArtifactMetadata) –

  The artifact metadata object to write to the cache.

Source code in src/mender_docker_lifecycle_helper/helper.py

def update_cache_artifact_metadata(
    self, artifact_metadata: ArtifactMetadata
) -> None:
    """
    Write the specified artifact metadata to the cache previous metadata location, if applicable.

    :param artifact_metadata: The artifact metadata object to write to the cache.
    """
    if self.context.cache:
        self.context.logger.debug(
            f"Updating cached metadata at {self.context.cache_artifact_metadata_file}"
        )
        artifact_metadata.to_file(self.context.cache_artifact_metadata_file)

mender_docker_lifecycle_helper.utils.container_utils

Functions:

• get_image_hash –

  Read the image hash for a given image reference from a remote registry, or, if unavailable, from the local image store.

• save_image_to_file –

  Saves the specified container image to the specified file.

• save_local_image_to_file –

  Saves the specified container image from the local image store to the specified file.

• save_registry_image_to_file –

  Saves the specified container image from a registry to the specified file.

_image_ref_hash_or_tag

_image_ref_hash_or_tag(image_registry: str, image_tag: str, image_hash: str) -> str

Reconstructs an image ref using only the hash (no tag) if provided, or the tag ref otherwise.

Parameters:

• image_registry

  (str) –

  The registry portion of image ref.

• image_tag

  (str) –

  The tag portion of image ref.

• image_hash

  (str) –

  The hash portion of image ref.

Returns:

• str –

  The reconstructed image ref.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def _image_ref_hash_or_tag(
    image_registry: str,
    image_tag: str,
    image_hash: str,
) -> str:
    """
    Reconstructs an image ref using only the hash (no tag) if provided, or the tag ref otherwise.

    :param image_registry: The registry portion of image ref.
    :param image_tag: The tag portion of image ref.
    :param image_hash: The hash portion of image ref.
    :return: The reconstructed image ref.
    """
    if image_hash is not None:
        return f"{image_registry}{REF_HASH_SEPARATOR}{image_hash}"
    elif image_tag is not None:
        return f"{image_registry}{REF_TAG_SEPARATOR}{image_tag}"
    else:
        return image_registry

_split_image_ref

_split_image_ref(image_ref: str) -> tuple[str, str, str]

Split an image ref into its component parts, if present.

Parameters:

• image_ref

  (str) –

  The image ref to split.

Returns:

• tuple[str, str, str] –

  A tuple of the image registry, tag, and hash, or None if not specified.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def _split_image_ref(image_ref: str) -> tuple[str, str, str]:
    """
    Split an image ref into its component parts, if present.

    :param image_ref: The image ref to split.
    :return: A tuple of the image registry, tag, and hash, or None if not specified.
    """
    image_registry = None
    image_tag = None
    image_hash = None
    if REF_HASH_SEPARATOR in image_ref:
        # Split into parts before and after the hash prefix
        image_ref, image_hash = image_ref.split(REF_HASH_SEPARATOR, 1)
    if REF_TAG_SEPARATOR in image_ref:
        # Find the last colon
        last_colon = image_ref.rfind(REF_TAG_SEPARATOR)
        # Check if there is a / after the colon (would indicate it is part of a path or port)
        if "/" in image_ref[last_colon:]:
            image_registry = image_ref
        # If there is no / after the colon, it is a tag
        else:
            image_registry = image_ref[:last_colon]
            image_tag = image_ref[(last_colon + 1) :]
    else:
        image_registry = image_ref
    return image_registry, image_tag, image_hash

get_image_hash

get_image_hash(image_ref: str, logger: Logger) -> str

Read the image hash for a given image reference from a remote registry, or, if unavailable, from the local image store.

Parameters:

• image_ref

  (str) –

  The image ref for which to get the hash.

• logger

  (Logger) –

  The logger object to which to report.

Returns:

• str –

  The hash of the image.

Raises:

• ValueError –

  If the image hash cannot be retrieved.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def get_image_hash(
    image_ref: str,
    logger: logging.Logger,
) -> str:
    """
    Read the image hash for a given image reference from a remote registry, or, if unavailable, from the local image store.

    :param image_ref: The image ref for which to get the hash.
    :param logger: The logger object to which to report.
    :raises ValueError: If the image hash cannot be retrieved.
    :return: The hash of the image.
    """

    image_hash = None
    # Try docker buildx imagetools inspect for remote images
    try:
        image_hash = (
            subprocess.run(
                [
                    DOCKER_BIN,
                    "buildx",
                    "imagetools",
                    "inspect",
                    image_ref,
                    "--format",
                    '"{{json .Manifest.Digest}}"',
                ],
                capture_output=True,
                text=True,
                check=True,
            )
            .stdout.strip()
            .strip('"')
            .removeprefix(f"{HASH_PREFIX}:")
        )
    except subprocess.CalledProcessError as e:
        logger.debug(
            f"docker buildx imagetools inspect failed for image {image_ref}: {e}\n{e.stderr}"
        )
        # Try docker inspect for local images
        try:
            image_hash = (
                subprocess.run(
                    [DOCKER_BIN, "inspect", "--format", "{{.Id}}", image_ref],
                    capture_output=True,
                    text=True,
                    check=True,
                )
                .stdout.strip()
                .removeprefix(f"{HASH_PREFIX}:")
            )
        except subprocess.CalledProcessError as e:
            logger.debug(
                f"docker inspect failed for image {image_ref}: {e}\n{e.stderr}"
            )
            raise ValueError(f"Could not retrieve hash for image: {image_ref}")

    return image_hash

save_image_to_file

save_image_to_file(image: dict[str, str], file: Path, platform: str = None) -> CompletedProcess

Saves the specified container image to the specified file.

Parameters:

• image

  (dict[str, str]) –

  The metadata (as {ref: , hash: }) of the image to save.

• file

  (Path) –

  The path of the file to which to save the image.

• platform

  (str, default: None ) –

  The platform for which to save the image, as os/arch[/variant].

Returns:

• CompletedProcess –

  The completed process from the subprocess call.

Raises:

• ImageRefHashMismatchException –

  If the metadata ref contains a hash that does not match the provided metadata hash.

• ImageNotFoundException –

  If the image cannot be found.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def save_image_to_file(
    image: dict[str, str], file: Path, platform: str = None
) -> subprocess.CompletedProcess:
    """
    Saves the specified container image to the specified file.

    :param image: The metadata (as {ref: <ref>, hash: <hash>}) of the image to save.
    :param file: The path of the file to which to save the image.
    :param platform: The platform for which to save the image, as os/arch[/variant].
    :raises ImageRefHashMismatchException: If the metadata ref contains a hash that does not match the provided metadata hash.
    :raises ImageNotFoundException: If the image cannot be found.
    :return: The completed process from the subprocess call.
    """
    image_ref = image["ref"]
    image_hash = image["hash"]

    image_registry, image_tag, image_ref_hash = _split_image_ref(image_ref)
    if image_ref_hash is not None and image_ref_hash != image_hash:
        raise ImageRefHashMismatchException(
            f"Specified hash {image_hash} does not match hash embedded in ref {image_ref}."
        )
    image_ref = _image_ref_hash_or_tag(image_registry, image_tag, image_hash)

    try:
        return save_local_image_to_file(image_hash, file)
    except ImageNotFoundException:
        try:
            return save_registry_image_to_file(image_ref, file, platform)
        except ImageNotFoundException:
            raise ImageNotFoundException(
                f"Image with ref {image_ref} not found in local daemon or remote registry"
            )

save_local_image_to_file

save_local_image_to_file(image_hash: str, file: Path) -> CompletedProcess

Saves the specified container image from the local image store to the specified file.

Parameters:

• image_hash

  (str) –

  The hash of the image to save.

• file

  (Path) –

  The path of the file to which to save the image.

Returns:

• CompletedProcess –

  The completed process from the subprocess call.

Raises:

• ImageNotFoundException –

  If the save operation cannot find the image.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def save_local_image_to_file(
    image_hash: str, file: Path
) -> subprocess.CompletedProcess:
    """
    Saves the specified container image from the local image store to the specified file.

    :param image_hash: The hash of the image to save.
    :param file: The path of the file to which to save the image.
    :raises ImageNotFoundException: If the save operation cannot find the image.
    :return: The completed process from the subprocess call.
    """
    # skopeo copy docker-daemon:<image_hash> oci-archive:<file>
    try:
        result = subprocess.run(
            [
                SKOPEO_BIN,
                "copy",
                f"{DAEMON_TRANSPORT}{HASH_PREFIX}:{image_hash}",
                f"{OCI_TRANSPORT}:{file}",
            ],
            capture_output=True,
            check=True,
        )
        if result.returncode != 0:
            raise subprocess.SubprocessError(result.stdout, result.stderr)
        return result
    except subprocess.CalledProcessError as e:
        stderr_str = (
            e.stderr.decode("utf-8", errors="ignore")
            if isinstance(e.stderr, bytes)
            else e.stderr
        )
        if (
            REF_NOT_FOUND_DAEMON_LOG in stderr_str
            or HASH_NOT_FOUND_DAEMON_LOG in stderr_str
        ):
            raise ImageNotFoundException(
                f"Hash {image_hash} not found in local daemon."
            )
        else:
            raise e

save_registry_image_to_file

save_registry_image_to_file(image_ref: str, file: Path, platform: str = None) -> CompletedProcess

Saves the specified container image from a registry to the specified file.

Parameters:

• image_ref

  (str) –

  The ref of the image to save.

• file

  (Path) –

  The path of the file to which to save the image.

• platform

  (str, default: None ) –

  The platform for which to save the image, as os/arch[/variant].

Returns:

• CompletedProcess –

  The completed process from the subprocess call.

Raises:

• ImageNotFoundException –

  If the save operation cannot find the image.

Source code in src/mender_docker_lifecycle_helper/utils/container_utils.py

def save_registry_image_to_file(
    image_ref: str, file: Path, platform: str = None
) -> subprocess.CompletedProcess:
    """
    Saves the specified container image from a registry to the specified file.

    :param image_ref: The ref of the image to save.
    :param file: The path of the file to which to save the image.
    :param platform: The platform for which to save the image, as os/arch[/variant].
    :raises ImageNotFoundException: If the save operation cannot find the image.
    :return: The completed process from the subprocess call.
    """

    image_ref = _image_ref_hash_or_tag(*_split_image_ref(image_ref))

    platform_dict = {"os": None, "arch": None, "variant": None}
    if platform is not None:
        platform_values = platform.split("/")
        platform_fields = platform_dict.keys()
        platform_dict = dict(
            zip_longest(platform_fields, platform_values, fillvalue=None)
        )

    # skopeo copy docker://<image_ref> oci-archive:<file>
    try:
        cmd = [
            SKOPEO_BIN,
            "copy",
        ]
        if platform_dict["os"] is not None:
            cmd.extend(["--override-os", platform_dict["os"]])
        if platform_dict["arch"] is not None:
            cmd.extend(["--override-arch", platform_dict["arch"]])
        if platform_dict["variant"] is not None:
            cmd.extend(["--override-variant", platform_dict["variant"]])
        cmd.extend(
            [
                f"{REGISTRY_TRANSPORT}{image_ref}",
                f"{OCI_TRANSPORT}:{file}",
            ]
        )
        result = subprocess.run(
            cmd,
            capture_output=True,
            check=True,
        )
        if result.returncode != 0:
            raise subprocess.SubprocessError(result.stdout, result.stderr)
        return result
    except subprocess.CalledProcessError as e:
        stderr_str = (
            e.stderr.decode("utf-8", errors="ignore")
            if isinstance(e.stderr, bytes)
            else e.stderr
        )
        if REF_NOT_FOUND_REGISTRY_LOG in stderr_str:
            raise ImageNotFoundException(f"Ref {image_ref} not found in registry.")
        else:
            raise subprocess.SubprocessError(stderr_str)

mender_docker_lifecycle_helper.utils.deep_delta

Functions:

• oci_deep_delta –

  Generate deep delta between OCI images.

_read_layers_from_manifest

_read_layers_from_manifest(image_dir: Path, target_platform: str, logger: Logger | None = getLogger(__name__)) -> list[Path]

Reads the layers from an OCI image manifest file.

Parameters:

• image_dir

  (Path) –

  The directory in which the image is extracted.

• target_platform

  (str) –

  The platform to target, if the image contains multiple, as os/[architecture]/[variant].

• logger

  (Logger | None, default: getLogger(__name__) ) –

  A logger with which to log steps of function processes, defaults to logging.getLogger(name).

Returns:

• list[Path] –

  The list of paths to the layers referenced by the image manifest.

Source code in src/mender_docker_lifecycle_helper/utils/deep_delta.py

def _read_layers_from_manifest(
    image_dir: Path,
    target_platform: str,
    logger: logging.Logger | None = logging.getLogger(__name__),
) -> list[Path]:
    """
    Reads the layers from an OCI image manifest file.

    :param image_dir: The directory in which the image is extracted.
    :param target_platform: The platform to target, if the image contains multiple, as os/[architecture]/[variant].
    :param logger: A logger with which to log steps of function processes, defaults to logging.getLogger(__name__).
    :return: The list of paths to the layers referenced by the image manifest.
    """
    blobs_dir = image_dir / "blobs" / HASH_PREFIX

    index = {}
    index_filename = image_dir / "index.json"
    logger.debug(f"Reading the image index file {index_filename}.")
    with open(index_filename, "r") as index_file:
        index = json.load(index_file)

    manifest_hash = index["manifests"][0]["digest"].removeprefix(f"{HASH_PREFIX}:")
    manifest = {}
    manifest_filename = blobs_dir / manifest_hash
    logger.debug(f"Reading manifest file {manifest_filename}.")
    with open(blobs_dir / manifest_hash) as manifest_file:
        manifest = json.load(manifest_file)

    # Multi-platform images may have multiple levels of nested manifests.
    # Loop through manifests as needed until layers are found.
    visited = set()
    while "layers" not in manifest and "manifests" in manifest:
        logger.debug(
            "Manifest without layers detected, looking for matching platform manifest..."
        )
        found = False
        for platform_manifest in manifest["manifests"]:
            platform_values = target_platform.split("/")
            platform_fields = ["os", "architecture", "variant"]
            if all(
                platform_manifest["platform"].get(platform_field) == platform_value
                for platform_field, platform_value in zip(
                    platform_fields, platform_values
                )
            ):
                logger.debug("Matching platform found!")
                new_hash = platform_manifest["digest"].removeprefix(f"{HASH_PREFIX}:")
                if f"{blobs_dir}-{new_hash}" in visited:
                    raise ImageDeltaException("Cycle detected while resolving manifest")
                visited.add(f"{blobs_dir}-{new_hash}")
                with open(blobs_dir / new_hash) as manifest_file:
                    manifest = json.load(manifest_file)
                found = True
                break
        if not found:
            raise ImageDeltaException("No matching platform found in manifest")

    layers = [
        blobs_dir / layer["digest"].removeprefix(f"{HASH_PREFIX}:")
        for layer in manifest["layers"]
    ]
    return layers

oci_deep_delta

oci_deep_delta(from_dir: Path, to_dir: Path, delta_dir: Path, delta_filename: str, target_platform: str, delta_cmd: list[str] | None = XDELTA_CMD, logger: Logger | None = getLogger(__name__)) -> Path

Generate deep delta between OCI images.

Writes .vcdiff and .source files into delta_dir based on the layer files in to_dir and from_dir, bundled into a tar file.

Parameters:

• from_dir

  (Path) –

  The path to the extracted current image.

• to_dir

  (Path) –

  The path to the extracted new image.

• delta_dir

  (Path) –

  The path under which to create the delta layers.

• delta_filename

  (str) –

  The name of the delta image archive file to create.

• delta_cmd

  (list[str] | None, default: XDELTA_CMD ) –

  The command to use for layer delta generation, defaults to XDELTA_CMD.

• target_platform

  (str) –

  The platform to target, if the image contains multiple, as os/[architecture]/[variant].

• logger

  (Logger | None, default: getLogger(__name__) ) –

  A logger with which to log steps of function processes, defaults to logging.getLogger(name).

Returns:

• Path –

  The image delta file.

Raises:

• ImageDeltaException –

  If images contain different numbers of layers.

Source code in src/mender_docker_lifecycle_helper/utils/deep_delta.py

def oci_deep_delta(
    from_dir: Path,
    to_dir: Path,
    delta_dir: Path,
    delta_filename: str,
    target_platform: str,
    delta_cmd: list[str] | None = XDELTA_CMD,
    logger: logging.Logger | None = logging.getLogger(__name__),
) -> Path:
    """
    Generate deep delta between OCI images.

    Writes .vcdiff and .source files into delta_dir based on the layer files in to_dir and from_dir, bundled into a tar file.

    :param from_dir: The path to the extracted current image.
    :param to_dir: The path to the extracted new image.
    :param delta_dir: The path under which to create the delta layers.
    :param delta_filename: The name of the delta image archive file to create.
    :param delta_cmd: The command to use for layer delta generation, defaults to XDELTA_CMD.
    :param target_platform: The platform to target, if the image contains multiple, as os/[architecture]/[variant].
    :param logger: A logger with which to log steps of function processes, defaults to logging.getLogger(__name__).
    :raises ImageDeltaException: If images contain different numbers of layers.
    :return: The image delta file.
    """
    # Load layers from current image manifest
    from_layers = _read_layers_from_manifest(from_dir, target_platform, logger)

    # Load layers from new image manifest
    to_layers = _read_layers_from_manifest(to_dir, target_platform, logger)

    # Check if current image has more layers than new image
    if len(from_layers) > len(to_layers):
        logger.error(
            "Failed to create image delta because the source image has more layers than the new one."
        )
        raise ImageDeltaException(
            "Image delta generation failed: source image has more layers than the new one"
        )

    delta_gen_dir = delta_dir / "gen"
    logger.debug(f"Copying to-image dir {to_dir} to new delta dir {delta_gen_dir}.")
    delta_gen_dir.mkdir(parents=True)
    # Copy full dir to capture layout and accompanying files
    shutil.copytree(to_dir, delta_gen_dir, dirs_exist_ok=True)

    # Generate deltas for each layer pair
    for i in range(len(from_layers)):
        logger.debug(f"Diffing layer {i + 1} of {len(from_layers)}")
        from_layer_path = from_layers[i]
        to_layer_path = to_layers[i]
        to_layer = to_layer_path.name
        delta_layer_path = delta_gen_dir / "blobs" / HASH_PREFIX / to_layer
        source_path = delta_gen_dir / "blobs" / HASH_PREFIX / (to_layer + ".source")
        vcdiff_layer_path = (
            delta_gen_dir / "blobs" / HASH_PREFIX / (to_layer + ".vcdiff")
        )

        # Run delta command, creating the vcdiff file in the delta dir
        try:
            result = subprocess.run(
                [*delta_cmd, from_layer_path, to_layer_path, vcdiff_layer_path],
                capture_output=True,
                check=True,
            )
            if result.returncode != 0:
                raise subprocess.SubprocessError(result.stdout, result.stderr)
        except subprocess.CalledProcessError as e:
            stderr_str = (
                e.stderr.decode("utf-8", errors="ignore")
                if isinstance(e.stderr, bytes)
                else e.stderr
            )
            raise subprocess.SubprocessError(stderr_str)

        # Remove the layer file from the delta dir
        delta_layer_path.unlink()

        # Write source layer reference in the delta dir.
        # The .source file must contain the relative path from the image root to the source blob,
        # because the docker-compose module prepends current_image_dir to this value.
        source_relative_path = from_layers[i].relative_to(from_dir)
        with open(source_path, "w") as f:
            f.write(str(source_relative_path))

    delta_file = delta_dir / delta_filename
    logger.debug(f"Layer diffing complete, creating delta file {delta_file}...")
    with tarfile.open(delta_file, "w") as tar:
        tar.add(delta_gen_dir, arcname=".")
    logger.debug(
        f"Delta file {delta_file} complete. Cleaning up delta gen dir {delta_gen_dir}."
    )
    shutil.rmtree(delta_gen_dir)
    return delta_file

mender_docker_lifecycle_helper.utils.image_cache

Classes:

• ImageCache –

  Represents a cache of container images archive files, extractions of those archives, and files of the computed deltas between images. Images are referenced in the cache by manifest hash (and from/to hashes for deltas), and can be specified for inclusion in the cache by image ref and hash, or by OCI image filename.

• ImageDirFormatException –

  Raised when an OCI image directory is not in the expected format.

ImageCache

ImageCache(cache_dir: Path, platform: str | None = None, delta_cache_dirname: str | None = DELTA_CACHE_DIRNAME, extract_cache_dirname: str | None = EXTRACT_CACHE_DIRNAME, image_file_name: str | None = IMAGE_FILE_NAME, save_cache_dirname: str | None = SAVE_CACHE_DIRNAME, logger: Logger | None = getLogger(__name__))

Represents a cache of container images archive files, extractions of those archives, and files of the computed deltas between images. Images are referenced in the cache by manifest hash (and from/to hashes for deltas), and can be specified for inclusion in the cache by image ref and hash, or by OCI image filename.

Construct an ImageCache object, reading cache dir contents into maps if present or creating them as empty dirs.

Parameters:

• cache_dir

  (Path) –

  The top-level directory of the image cache.

• platform

  (str | None, default: None ) –

  The platform for which to save images, as os/arch[/variant].

• delta_cache_dirname

  (str | None, default: DELTA_CACHE_DIRNAME ) –

  The name of the subdirectory for image deltas, defaults to DELTA_CACHE_DIRNAME.

• extract_cache_dirname

  (str | None, default: EXTRACT_CACHE_DIRNAME ) –

  The name of the subdirectory for image extracts, defaults to EXTRACT_CACHE_DIRNAME.

• image_file_name

  (str | None, default: IMAGE_FILE_NAME ) –

  The filename to use for image save files, defaults to IMAGE_FILE_NAME.

• save_cache_dirname

  (str | None, default: SAVE_CACHE_DIRNAME ) –

  The name of the subdirectory for image saves, defaults to SAVE_CACHE_DIRNAME.

• logger

  (Logger | None, default: getLogger(__name__) ) –

  A logger with which to log steps of cache processes, defaults to logging.getLogger(name).

Methods:

• cleanup_by_mtime –

  Remove oldest cached items to bring cache size at or below limit.

• delta –

  Get the delta file path for a given from and to image hash, creating folders if required.

• extract_cache_file –

  Extract image from a specified file to the cache and read and return the image metadata.

• extract_cache_image –

  Extract an image with the given hash in the cache and return the path to the extracted image dir. If the image is not yet in the image save cache, it will be added there first.

• save_cache_image –

  Save an image with the given ref and hash to the save cache, if not already present.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def __init__(
    self,
    cache_dir: Path,
    platform: str | None = None,
    delta_cache_dirname: str | None = DELTA_CACHE_DIRNAME,
    extract_cache_dirname: str | None = EXTRACT_CACHE_DIRNAME,
    image_file_name: str | None = IMAGE_FILE_NAME,
    save_cache_dirname: str | None = SAVE_CACHE_DIRNAME,
    logger: logging.Logger | None = logging.getLogger(__name__),
):
    """
    Construct an ImageCache object, reading cache dir contents into maps if present or creating them as empty dirs.

    :param cache_dir: The top-level directory of the image cache.
    :param platform: The platform for which to save images, as os/arch[/variant].
    :param delta_cache_dirname: The name of the subdirectory for image deltas, defaults to DELTA_CACHE_DIRNAME.
    :param extract_cache_dirname: The name of the subdirectory for image extracts, defaults to EXTRACT_CACHE_DIRNAME.
    :param image_file_name: The filename to use for image save files, defaults to IMAGE_FILE_NAME.
    :param save_cache_dirname: The name of the subdirectory for image saves, defaults to SAVE_CACHE_DIRNAME.
    :param logger: A logger with which to log steps of cache processes, defaults to logging.getLogger(__name__).
    """
    self.logger = logger
    self.image_file_name = image_file_name
    self.platform = platform
    self.cache_dir = cache_dir

    cache_dir.mkdir(parents=True, exist_ok=True)
    self.delta_cache_dir = cache_dir / delta_cache_dirname
    self.delta_cache_dir.mkdir(exist_ok=True)
    self.delta_cache = {
        from_dir.name: {
            to_dir.name: to_dir / image_file_name
            for to_dir in from_dir.iterdir()
            if to_dir.is_dir()
        }
        for from_dir in self.delta_cache_dir.iterdir()
        if from_dir.is_dir()
    }
    self.extract_cache_dir = cache_dir / extract_cache_dirname
    self.extract_cache_dir.mkdir(exist_ok=True)
    self.extract_cache = {
        hash_dir.name: hash_dir
        for hash_dir in self.extract_cache_dir.iterdir()
        if hash_dir.is_dir()
    }
    self.save_cache_dir = cache_dir / save_cache_dirname
    self.save_cache_dir.mkdir(exist_ok=True)
    self.save_cache = {
        hash_dir.name: hash_dir / image_file_name
        for hash_dir in self.save_cache_dir.iterdir()
        if hash_dir.is_dir()
    }

_calculate_space_to_free_for_percent

_calculate_space_to_free_for_percent(disk_percent: float, current_cache_size: int) -> int

Calculate how much space needs to be freed to meet the disk percent threshold.

Parameters:

• disk_percent

  (float) –

  Minimum percent of total disk that should remain free.

• current_cache_size

  (int) –

  Current cache size in bytes.

Returns:

• int –

  Bytes to free, or 0 if no cleanup needed or cache is already within limits.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _calculate_space_to_free_for_percent(
    self, disk_percent: float, current_cache_size: int
) -> int:
    """
    Calculate how much space needs to be freed to meet the disk percent threshold.

    :param disk_percent: Minimum percent of total disk that should remain free.
    :param current_cache_size: Current cache size in bytes.
    :return: Bytes to free, or 0 if no cleanup needed or cache is already within limits.
    """
    stats = self._get_disk_stats()
    min_free_bytes = int(stats["total"] * disk_percent / 100)
    current_free = stats["free"]
    additional_needed = min_free_bytes - current_free

    if additional_needed <= 0:
        return 0

    # We can only free up to the current cache size
    return min(additional_needed, current_cache_size)

_extract_oci_file

_extract_oci_file(extract_dir: Path, extract_file: Path) -> None

Extract specified OCI image file into specified directory, and ensure OCI-compliant layout.

Parameters:

• extract_dir

  (Path) –

  The path into which to extract the OCI file.

• extract_file

  (Path) –

  The OCI file to extract.

Returns:

• None –

  None.

Raises:

• ImageDirFormatException –

  Indicates that the specified image file is not in the correct format.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _extract_oci_file(self, extract_dir: Path, extract_file: Path) -> None:
    """
    Extract specified OCI image file into specified directory, and ensure OCI-compliant layout.

    :param extract_dir: The path into which to extract the OCI file.
    :param extract_file: The OCI file to extract.
    :raises ImageDirFormatException: Indicates that the specified image file is not in the correct format.
    :return: None.
    """
    with tarfile.open(extract_file, "r:*") as tar:
        tar.extractall(
            path=extract_dir,
            filter="tar",
        )

    if not (extract_dir / OCI_LAYOUT_FILENAME).exists():
        raise ImageDirFormatException(
            f"{extract_file} as extracted to {extract_dir} is not in valid OCI format."
        )

_get_cache_items_by_mtime

_get_cache_items_by_mtime() -> list[tuple[Path, float, int]]

Get all cache items ordered by modification time.

Returns:

• list[tuple[Path, float, int]] –

  List of (path, mtime, size) tuples. Items are always directories: save/, extract/, delta//.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _get_cache_items_by_mtime(self) -> list[tuple[Path, float, int]]:
    """
    Get all cache items ordered by modification time.

    :return: List of (path, mtime, size) tuples. Items are always directories:
        save/<hash>, extract/<hash>, delta/<from>/<to>.
    """
    items = []
    for hash_dir in self.save_cache_dir.iterdir():
        if hash_dir.is_dir():
            mtime = hash_dir.stat().st_mtime
            size = self._get_dir_size(hash_dir)
            items.append((hash_dir, mtime, size))

    for hash_dir in self.extract_cache_dir.iterdir():
        if hash_dir.is_dir():
            mtime = hash_dir.stat().st_mtime
            size = self._get_dir_size(hash_dir)
            items.append((hash_dir, mtime, size))

    for from_dir in self.delta_cache_dir.iterdir():
        if from_dir.is_dir():
            for to_dir in from_dir.iterdir():
                if to_dir.is_dir():
                    mtime = to_dir.stat().st_mtime
                    size = self._get_dir_size(to_dir)
                    items.append((to_dir, mtime, size))

    return sorted(items, key=lambda x: x[1])

_get_cache_size

_get_cache_size() -> int

Calculate total size of the images cache directory.

Returns:

• int –

  Total size in bytes.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _get_cache_size(self) -> int:
    """
    Calculate total size of the images cache directory.

    :return: Total size in bytes.
    """
    total = 0
    for subdir in [
        self.save_cache_dir,
        self.extract_cache_dir,
        self.delta_cache_dir,
    ]:
        for path in subdir.rglob("*"):
            if path.is_file():
                total += path.stat().st_size
    return total

_get_dir_size

_get_dir_size(path: Path) -> int

Recursively calculate total size of directory.

Parameters:

• path

  (Path) –

  The directory path to calculate size for.

Returns:

• int –

  Total size in bytes.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _get_dir_size(self, path: Path) -> int:
    """
    Recursively calculate total size of directory.

    :param path: The directory path to calculate size for.
    :return: Total size in bytes.
    """
    return sum(f.stat().st_size for f in path.rglob("*") if f.is_file())

_get_disk_stats

_get_disk_stats() -> dict

Get disk usage stats for the filesystem containing the cache.

Returns:

• dict –

  Dictionary with 'total' and 'free' keys in bytes.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _get_disk_stats(self) -> dict:
    """
    Get disk usage stats for the filesystem containing the cache.

    :return: Dictionary with 'total' and 'free' keys in bytes.
    """
    stat = os.statvfs(self.cache_dir)
    return {
        "total": stat.f_blocks * stat.f_frsize,
        "free": stat.f_bavail * stat.f_frsize,
    }

_sync_cache_dicts_on_cleanup

_sync_cache_dicts_on_cleanup() -> None

Remove entries from in-memory cache dicts after cleanup removes those dirs.

Returns:

• None –

  None.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def _sync_cache_dicts_on_cleanup(self) -> None:
    """
    Remove entries from in-memory cache dicts after cleanup removes those dirs.

    :return: None.
    """
    for hash_dir in list(self.save_cache.keys()):
        if not (self.save_cache_dir / hash_dir).exists():
            del self.save_cache[hash_dir]

    for hash_dir in list(self.extract_cache.keys()):
        if not (self.extract_cache_dir / hash_dir).exists():
            del self.extract_cache[hash_dir]

    for from_hash in list(self.delta_cache.keys()):
        for to_hash in list(self.delta_cache[from_hash].keys()):
            delta_path = self.delta_cache_dir / from_hash / to_hash
            if not delta_path.exists():
                del self.delta_cache[from_hash][to_hash]
        if not self.delta_cache[from_hash]:
            del self.delta_cache[from_hash]

cleanup_by_mtime

cleanup_by_mtime(limit_size_bytes: int | None = None, disk_percent: float | None = None) -> int

Remove oldest cached items to bring cache size at or below limit.

Only folder items (save/, extract/, delta//) are removed.

Parameters:

• limit_size_bytes

  (int | None, default: None ) –

  Maximum cache size in bytes. If None with disk_percent, no cleanup occurs.

• disk_percent

  (float | None, default: None ) –

  If set, calculate free space needed as percent of total disk.

Returns:

• int –

  Total bytes freed by cleanup.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def cleanup_by_mtime(
    self,
    limit_size_bytes: int | None = None,
    disk_percent: float | None = None,
) -> int:
    """
    Remove oldest cached items to bring cache size at or below limit.

    Only folder items (save/<hash>, extract/<hash>, delta/<from>/<to>) are removed.

    :param limit_size_bytes: Maximum cache size in bytes. If None with disk_percent,
        no cleanup occurs.
    :param disk_percent: If set, calculate free space needed as percent of total disk.
    :return: Total bytes freed by cleanup.
    """
    current_size = self._get_cache_size()
    bytes_freed = 0

    if disk_percent is not None:
        space_to_free = self._calculate_space_to_free_for_percent(
            disk_percent, current_size
        )
    elif limit_size_bytes is not None:
        space_to_free = current_size - limit_size_bytes
    else:
        self.logger.debug("No cleanup limit specified, skipping cleanup.")
        return 0

    if space_to_free <= 0:
        self.logger.debug("Cache size already within limits, skipping cleanup.")
        return 0

    self.logger.info(
        f"Cache cleanup triggered, need to free {space_to_free} bytes."
    )

    items = self._get_cache_items_by_mtime()
    for item_path, mtime, size in items:
        if space_to_free <= 0:
            break

        self.logger.debug(f"Removing cache item {item_path} ({size} bytes).")
        try:
            shutil.rmtree(item_path)
            bytes_freed += size
            space_to_free -= size
            self.logger.info(
                f"Removed {item_path}, freed {bytes_freed} bytes so far."
            )
        except Exception as e:
            self.logger.warning(f"Failed to remove {item_path}: {e}")

    self._sync_cache_dicts_on_cleanup()
    return bytes_freed

delta

delta(from_image: dict[str, str], to_image: dict[str, str]) -> Path

Get the delta file path for a given from and to image hash, creating folders if required.

Parameters:

• from_image

  (dict[str, str]) –

  The metadata (specifically {ref: , hash: }) of the image from which the delta is defined.

• to_image

  (dict[str, str]) –

  The metadata (specifically {ref: , hash: }) of the image to which the delta is defined.

Returns:

• Path –

  The path to the delta file for the given from and to image hash.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def delta(self, from_image: dict[str, str], to_image: dict[str, str]) -> Path:
    """
    Get the delta file path for a given from and to image hash, creating folders if required.

    :param from_image: The metadata (specifically {ref: <ref>, hash: <hash>}) of the image from which the delta is defined.
    :param to_image: The metadata (specifically {ref: <ref>, hash: <hash>}) of the image to which the delta is defined.
    :return: The path to the delta file for the given from and to image hash.
    """
    from_hash = from_image["hash"]
    to_hash = to_image["hash"]

    if to_hash in self.delta_cache.get(from_hash, {}):
        delta_file = self.delta_cache[from_hash][to_hash]
        self.logger.debug(
            f"Found cached delta file {delta_file} for from hash {from_hash} to hash {to_hash}."
        )
        # Update the file timestamp for cache cleanup logic
        delta_file.touch()
        return delta_file
    else:
        delta_dir = self.delta_cache_dir / from_hash / to_hash
        self.logger.debug(
            f"Creating delta for {from_hash} to {to_hash} under {delta_dir}..."
        )
        delta_file = oci_deep_delta(
            self.extract_cache_image(from_image),
            self.extract_cache_image(to_image),
            delta_dir,
            self.image_file_name,
            self.platform,
            logger=self.logger,
        )
        if from_hash not in self.delta_cache:
            self.delta_cache[from_hash] = {}
        self.delta_cache[from_hash][to_hash] = delta_file

        return delta_file

extract_cache_file

extract_cache_file(extract_file: Path) -> dict[str, str]

Extract image from a specified file to the cache and read and return the image metadata.

Parameters:

• extract_file

  (Path) –

  The path to the OCI image archive file to extract into the cache space.

Returns:

• dict[str, str] –

  The image specification (as {ref: , hash: }) as read from the file.

Raises:

• ImageDirFormatException –

  Indicates that the specified image file is not in the correct format.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def extract_cache_file(self, extract_file: Path) -> dict[str, str]:
    """
    Extract image from a specified file to the cache and read and return the image metadata.

    :param extract_file: The path to the OCI image archive file to extract into the cache space.
    :raises ImageDirFormatException: Indicates that the specified image file is not in the correct format.
    :return: The image specification (as {ref: <ref>, hash: <hash>}) as read from the file.
    """

    image_ref = ""
    image_hash = ""
    with tempfile.TemporaryDirectory(
        dir=self.extract_cache_dir
    ) as temp_extract_dir:
        temp_extract_dir = Path(temp_extract_dir)
        self.logger.debug(
            f"Extracting {extract_file} into temp dir {temp_extract_dir}"
        )
        self._extract_oci_file(temp_extract_dir, extract_file)

        image_index = {}
        with open(temp_extract_dir / OCI_INDEX_FILENAME) as index_file:
            image_index = json.load(index_file)

        image_manifest = image_index.get("manifests", [{}])[0]
        # Check index annotations first (for index-descriptor annotations from buildx)
        # then fall back to manifest annotations
        image_ref = image_index.get("annotations", {}).get(
            "io.containerd.image.name", None
        )
        if image_ref is None:
            image_ref = image_manifest.get("annotations", {}).get(
                "io.containerd.image.name", None
            )
        image_hash = image_manifest.get("digest", None)
        if image_hash is not None:
            image_hash = image_hash.removeprefix(f"{HASH_PREFIX}:")
        if image_ref is None:
            raise ImageDirFormatException(
                f"{extract_file} as extracted to {temp_extract_dir} does not contain expected io.containerd.image.name metadata in its index."
            )
        if image_hash is None:
            raise ImageDirFormatException(
                f"{extract_file} as extracted to {temp_extract_dir} does not contain digest metadata in its index."
            )

        image_extract_cache_dir = self.extract_cache_dir / image_hash
        self.logger.debug(
            f"Saving extracted image contents from {extract_file} to cache {image_extract_cache_dir}"
        )
        shutil.move(temp_extract_dir, image_extract_cache_dir)
        self.extract_cache[image_hash] = image_extract_cache_dir

    image_save_cache_folder = self.save_cache_dir / image_hash
    image_save_cache_folder.mkdir(parents=True, exist_ok=True)
    image_save_cache_file = image_save_cache_folder / self.image_file_name
    self.logger.debug(
        f"Saving image file {extract_file} to cache {image_save_cache_file}"
    )
    shutil.copy(extract_file, image_save_cache_file)
    self.save_cache[image_hash] = image_save_cache_file
    return {
        "ref": image_ref,
        "hash": image_hash,
    }

extract_cache_image

extract_cache_image(image: dict[str, str]) -> Path

Extract an image with the given hash in the cache and return the path to the extracted image dir. If the image is not yet in the image save cache, it will be added there first.

Parameters:

• image

  (dict[str, str]) –

  The metadata (specifically {ref: , hash: }) of the image to extract.

Returns:

• Path –

  The path to the extracted image directory.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def extract_cache_image(self, image: dict[str, str]) -> Path:
    """
    Extract an image with the given hash in the cache and return the path to the extracted image dir. If the image is not yet in the image save cache, it will be added there first.

    :param image: The metadata (specifically {ref: <ref>, hash: <hash>}) of the image to extract.
    :return: The path to the extracted image directory.
    """
    image_hash = image["hash"]
    if image_hash in self.extract_cache:
        self.logger.debug(
            f"Image {image['ref']} with hash {image_hash} already saved in cache at {self.extract_cache[image_hash]}."
        )
        return self.extract_cache[image_hash]

    image_file = self.save_cache_image(image)
    self.logger.debug(
        f"Extracting image file {image_file} with hash {image_hash} to cache."
    )
    extract_dir = self.extract_cache_dir / image_hash
    extract_dir.mkdir()
    self._extract_oci_file(extract_dir, image_file)
    self.extract_cache[image_hash] = extract_dir
    return extract_dir

save_cache_image

save_cache_image(image: dict[str, str]) -> Path

Save an image with the given ref and hash to the save cache, if not already present.

Parameters:

• image

  (dict[str, str]) –

  The metadata (specifically {ref: , hash: }) of the image to save.

Returns:

• Path –

  The path to the image file in the save cache.

Source code in src/mender_docker_lifecycle_helper/utils/image_cache.py

def save_cache_image(self, image: dict[str, str]) -> Path:
    """
    Save an image with the given ref and hash to the save cache, if not already present.

    :param image: The metadata (specifically {ref: <ref>, hash: <hash>}) of the image to save.
    :return: The path to the image file in the save cache.
    """
    image_hash = image["hash"]
    if image_hash in self.save_cache:
        save_image_file = self.save_cache[image_hash]
        self.logger.debug(
            f"Image with hash {image_hash} already saved in cache at {save_image_file}."
        )
        # Update the file timestamp for cache cleanup logic
        save_image_file.touch()
        return save_image_file
    else:
        save_image_dir = self.save_cache_dir / image_hash
        self.logger.debug(
            f"Saving image with hash {image_hash} for ref {image['ref']} to cache at {save_image_dir}."
        )
        save_image_dir.mkdir(parents=True, exist_ok=True)
        save_image_file = save_image_dir / self.image_file_name
        save_image_to_file(image, save_image_file, self.platform)
        self.logger.debug(f"Image file {save_image_file} saved.")
        self.save_cache[image_hash] = save_image_file
        return save_image_file

ImageDirFormatException

              flowchart TD
              mender_docker_lifecycle_helper.utils.image_cache.ImageDirFormatException[ImageDirFormatException]

              

              click mender_docker_lifecycle_helper.utils.image_cache.ImageDirFormatException href "" "mender_docker_lifecycle_helper.utils.image_cache.ImageDirFormatException"
            

Raised when an OCI image directory is not in the expected format.

mender_docker_lifecycle_helper.utils.mender_server

Functions:

• call_mender_host_api –

  Calls the Mender server API at the specified endpoint with provided args.

• get_deployment_status –

  Get the status of a deployment from the Mender server.

• upload_artifact –

  Upload the specified artifact file to the Mender server with retry logic.

• wait_for_deployment –

  Watch a deployment until completion, success, or timeout.

call_mender_host_api

call_mender_host_api(context: LifecycleHelperContext, mender_endpoint: str, request_args: dict, max_retries: int = 0, base_delay: int = 5) -> Response

Calls the Mender server API at the specified endpoint with provided args.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• mender_endpoint

  (str) –

  The endpoint of the Mender server to call.

• request_args

  (dict) –

  The args to provide to the API call.

• max_retries

  (int, default: 0 ) –

  Maximum number of retries for server errors (5xx). Default 0 (no retries).

• base_delay

  (int, default: 5 ) –

  Base delay in seconds between retries. Default 5.

Returns:

• Response –

  The request response object or None.

Raises:

• HTTPError –

  If the API call fails.

Source code in src/mender_docker_lifecycle_helper/utils/mender_server.py

def call_mender_host_api(
    context: LifecycleHelperContext,
    mender_endpoint: str,
    request_args: dict,
    max_retries: int = 0,
    base_delay: int = 5,
) -> requests.Response:
    """
    Calls the Mender server API at the specified endpoint with provided args.

    :param context: The context of the lifecycle helper execution.
    :param mender_endpoint: The endpoint of the Mender server to call.
    :param request_args: The args to provide to the API call.
    :param max_retries: Maximum number of retries for server errors (5xx). Default 0 (no retries).
    :param base_delay: Base delay in seconds between retries. Default 5.
    :raises HTTPError: If the API call fails.
    :return: The request response object or None.
    """
    if context.mender_pat is None:
        context.logger.error(
            "No MENDER_PAT env var specified, will not upload or deploy to the Mender server."
        )
        return None

    for attempt in range(max_retries + 1):
        r = requests.post(
            f"{context.mender_host}/api/management/v1/{mender_endpoint}",
            headers={
                "Accept": "application/json",
                "Authorization": f"Bearer {context.mender_pat}",
            },
            **request_args,
        )
        if r.status_code == 201:
            return r
        context.logger.error(
            f"Request failed for endpoint {mender_endpoint}: status={r.status_code}, text={r.text}, url={r.request.url}, headers={r.request.headers}"
        )
        if r.status_code >= 500 and attempt < max_retries:
            delay = base_delay * (2**attempt)
            context.logger.warning(
                f"Request attempt {attempt + 1} failed with "
                f"status {r.status_code}, retrying in {delay}s"
            )
            time.sleep(delay)
        else:
            r.raise_for_status()

get_deployment_status

get_deployment_status(context: LifecycleHelperContext, deployment_id: str) -> dict

Get the status of a deployment from the Mender server.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• deployment_id

  (str) –

  The ID of the deployment to check.

Returns:

• dict –

  The deployment statistics as a dict, or None if the request fails.

Source code in src/mender_docker_lifecycle_helper/utils/mender_server.py

def get_deployment_status(
    context: LifecycleHelperContext,
    deployment_id: str,
) -> dict:
    """
    Get the status of a deployment from the Mender server.

    :param context: The context of the lifecycle helper execution.
    :param deployment_id: The ID of the deployment to check.
    :return: The deployment statistics as a dict, or None if the request fails.
    """
    if context.mender_pat is None:
        context.logger.error(
            "No MENDER_PAT env var specified, cannot check deployment status."
        )
        return None

    r = requests.get(
        f"{context.mender_host}/api/management/v1/deployments/deployments/{deployment_id}/statistics",
        headers={
            "Accept": "application/json",
            "Authorization": f"Bearer {context.mender_pat}",
        },
    )
    if r.status_code != 200:
        context.logger.error(
            f"Failed to get deployment status: status={r.status_code}, text={r.text}"
        )
        r.raise_for_status()
    return r.json()

upload_artifact

upload_artifact(context: LifecycleHelperContext, artifact: LifecycleHelperArtifact, max_retries: int = 3, base_delay: int = 5) -> None

Upload the specified artifact file to the Mender server with retry logic.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• artifact

  (LifecycleHelperArtifact) –

  The object of the artifact to upload to the Mender server.

• max_retries

  (int, default: 3 ) –

  The maximum number of times to retry uploading the artifact in the case of failure.

• base_delay

  (int, default: 5 ) –

  The number of seconds to wait between upload retries.

Returns:

• None –

  None

Source code in src/mender_docker_lifecycle_helper/utils/mender_server.py

def upload_artifact(
    context: LifecycleHelperContext,
    artifact: LifecycleHelperArtifact,
    max_retries: int = 3,
    base_delay: int = 5,
) -> None:
    """
    Upload the specified artifact file to the Mender server with retry logic.

    :param context: The context of the lifecycle helper execution.
    :param artifact: The object of the artifact to upload to the Mender server.
    :param max_retries: The maximum number of times to retry uploading the artifact in the case of failure.
    :param base_delay: The number of seconds to wait between upload retries.
    :return: None
    """
    with open(artifact.filename, "rb") as file_contents:
        call_mender_host_api(
            context,
            "deployments/artifacts",
            {
                "data": {
                    "size": artifact.filename.stat().st_size,
                    "description": "string",
                },
                "files": {"artifact": file_contents},
            },
            max_retries=max_retries,
            base_delay=base_delay,
        )
    context.logger.info(f"Uploaded artifact {artifact.filename}")

wait_for_deployment

wait_for_deployment(context: LifecycleHelperContext, deployment_id: str, poll_interval: int = 30, timeout: int = 3600) -> bool

Watch a deployment until completion, success, or timeout.

Returns True if the deployment succeeded (all devices reported success), False otherwise.

Parameters:

• context

  (LifecycleHelperContext) –

  The context of the lifecycle helper execution.

• deployment_id

  (str) –

  The ID of the deployment to watch.

• poll_interval

  (int, default: 30 ) –

  Seconds between status checks (default: 30).

• timeout

  (int, default: 3600 ) –

  Maximum seconds to wait (default: 3600).

Returns:

• bool –

  True if deployment succeeded, False otherwise.

Source code in src/mender_docker_lifecycle_helper/utils/mender_server.py

def wait_for_deployment(
    context: LifecycleHelperContext,
    deployment_id: str,
    poll_interval: int = 30,
    timeout: int = 3600,
) -> bool:
    """
    Watch a deployment until completion, success, or timeout.

    Returns True if the deployment succeeded (all devices reported success),
    False otherwise.

    :param context: The context of the lifecycle helper execution.
    :param deployment_id: The ID of the deployment to watch.
    :param poll_interval: Seconds between status checks (default: 30).
    :param timeout: Maximum seconds to wait (default: 3600).
    :return: True if deployment succeeded, False otherwise.
    """
    start_time = time.time()

    while time.time() - start_time < timeout:
        stats = get_deployment_status(context, deployment_id)
        if stats is None:
            context.logger.error("Failed to get deployment status.")
            return False

        # stats contains: success, failure, pending, installing counts
        # and status string ("finished", "inprogress", etc.)
        context.logger.info(
            f"Deployment {deployment_id} status: "
            f"success={stats.get('success', 0)}, "
            f"failure={stats.get('failure', 0)}, "
            f"pending={stats.get('pending', 0)}, "
            f"installing={stats.get('installing', 0)}"
        )

        # Check if deployment is complete (no more pending or installing)
        total_active = stats.get("pending", 0) + stats.get("installing", 0)
        if total_active == 0:
            success_count = stats.get("success", 0)
            failure_count = stats.get("failure", 0)
            if failure_count > 0:
                context.logger.error(
                    f"Deployment {deployment_id} failed: "
                    f"{failure_count} device(s) reported failure."
                )
                return False
            if success_count > 0:
                context.logger.info(
                    f"Deployment {deployment_id} succeeded: "
                    f"{success_count} device(s) reported success."
                )
                return True
            context.logger.debug(f"Deployment {deployment_id} has no results yet.")

        context.logger.debug(f"Waiting {poll_interval}s before next status check...")
        time.sleep(poll_interval)

    context.logger.error(
        f"Deployment {deployment_id} timed out after {timeout} seconds."
    )
    return False