Skip to content

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: Used by:

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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@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 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
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:

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:

  • ManifestContentMismatchException –

    If service name in args not found in manifest.

Source code in src/mender_docker_lifecycle_helper/artifact.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@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 ManifestContentMismatchException: 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 ManifestContentMismatchException(
                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 ManifestContentMismatchException(
                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 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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, self.context.platform
    )
    (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 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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}"
    )
    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 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
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(version: str, services: Optional[dict[str, dict[str, dict[str, str]]]])

Construct an ArtifactMetadata object.

Parameters:

  • version

    (str) –

    The version identifier for the artifact.

  • services

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

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

Returned by: Used by:

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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(
    self, version: str, services: Optional[dict[str, dict[str, dict[str, str]]]]
):
    """
    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 
28
29
30
31
32
33
34
35
36
@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 
38
39
40
41
42
43
44
45
46
47
48
@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 
50
51
52
53
54
55
56
57
58
59
60
61
62
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 
64
65
66
67
68
69
70
71
72
73
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.

Parameters:

  • args

    –

    An object of CLI args for the execution as prepared by Click decorators.

Returns:

  • None –

    None

Source code in src/mender_docker_lifecycle_helper/cli.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@click.command(context_settings={"show_default": True})
@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(
    "--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=True,
    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=True,
    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(
    "-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,
    ),
)
@click.version_option()
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.

    :param args: An object of CLI args for the execution as prepared by Click decorators.
    :return: None
    """
    args = SimpleNamespace(**args)

    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

    LifecycleHelper(args).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:

Methods:

  • 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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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.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.logger = self._prep_logger(args.log_level)

    self.manifest_file = args.manifest_file.resolve()
    with open(self.manifest_file, "r") as f:
        self.manifest = yaml.safe_load(f.read())

    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)
        self.temp_dir = self.cache_dir / "temp"
        self.temp_dir.mkdir()
        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"
        )
        self.image_cache = ImageCache(self.cache_dir / "images")

    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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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
                }
            }
        }
    """
    with open(compose_file, "r") as f:
        compose = yaml.safe_load(f.read())

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

_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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@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 
156
157
158
159
160
161
162
163
164
165
166
167
168
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 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@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()
    formatter = logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.setLevel(getattr(logging, log_level))
    return logger

_prep_previous_artifact_metadata 

_prep_previous_artifact_metadata(previous_version: Optional[str]) -> 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
    (Optional[str]) –

    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 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def _prep_previous_artifact_metadata(
    self,
    previous_version: Optional[str],
) -> 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 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@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 
144
145
146
147
148
149
150
151
152
153
154
@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 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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

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 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
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 

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.

  • upload_artifact –

    Upload the specified artifact file to the Mender server.

Source code in src/mender_docker_lifecycle_helper/helper.py 
18
19
20
21
22
23
24
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:

Source code in src/mender_docker_lifecycle_helper/helper.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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) -> None

Issue a deployment of a pre-uploaded artifact.

Parameters:

Returns:

  • None –

    None

Source code in src/mender_docker_lifecycle_helper/helper.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def deploy_artifact(self, artifact: LifecycleHelperArtifact) -> None:
    """
    Issue a deployment of a pre-uploaded artifact.

    :param artifact: The object of the artifact to deploy to the Mender server.
    :return: None
    """
    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}"
    )
    call_mender_host_api(
        self.context,
        f"deployments/deployments/group/{self.context.device_group}",
        {
            "json": {
                "name": deployment_name,
                "artifact_name": artifact.name,
            }
        },
    )
    self.context.logger.info(f"Created deployment {deployment_name}")

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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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)
    self.upload_artifact(artifact)

    if self.context.device_group is not None:
        self.deploy_artifact(artifact)
        self.context.logger.debug(
            f"Artifact {artifact.name} deployed; updating cached metadata at {self.context.cache_artifact_metadata_file}"
        )
        artifact_metadata.to_file(self.context.cache_artifact_metadata_file)
    else:
        self.context.logger.debug(
            "No device group set; skipping deployment creation."
        )

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

upload_artifact 

upload_artifact(artifact: LifecycleHelperArtifact) -> None

Upload the specified artifact file to the Mender server.

Parameters:

Returns:

  • None –

    None

Source code in src/mender_docker_lifecycle_helper/helper.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def upload_artifact(self, artifact: LifecycleHelperArtifact) -> None:
    """
    Upload the specified artifact file to the Mender server.

    :param artifact: The object of the artifact to upload to the Mender server.
    :return: None
    """
    with open(artifact.filename, "rb") as file_contents:
        call_mender_host_api(
            self.context,
            "deployments/artifacts",
            {
                "data": {
                    "size": artifact.filename.stat().st_size,
                    "description": "string",
                },
                "files": {"artifact": file_contents},
            },
        )
        self.context.logger.info(f"Uploaded artifact {artifact.filename}")

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 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
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) -> 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.

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 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def save_image_to_file(
    image: dict[str, str], file: Path
) -> 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.
    :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)
        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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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) -> 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.

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 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def save_registry_image_to_file(
    image_ref: str, file: Path
) -> 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.
    :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))
    # skopeo copy docker://<image_ref> oci-archive:<file>
    try:
        result = subprocess.run(
            [
                SKOPEO_BIN,
                "copy",
                f"{REGISTRY_TRANSPORT}{image_ref}",
                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_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:

_read_layers_from_manifest 

_read_layers_from_manifest(image_dir: Path, target_platform: str, logger: Optional[Logger] = 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

    (Optional[Logger], 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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def _read_layers_from_manifest(
    image_dir: Path,
    target_platform: str,
    logger: Optional[logging.Logger] = 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 manifests from the index.
    if "layers" not in manifest and "manifests" in manifest:
        logger.debug(
            "Multi-platform image detected, looking for matching platform manifest..."
        )
        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!")
                with open(
                    blobs_dir
                    / platform_manifest["digest"].removeprefix(f"{HASH_PREFIX}:")
                ) as manifest_file:
                    manifest = json.load(manifest_file)

    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: Optional[list[str]] = XDELTA_CMD, logger: Optional[Logger] = 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

    (Optional[list[str]], 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

    (Optional[Logger], 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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def oci_deep_delta(
    from_dir: Path,
    to_dir: Path,
    delta_dir: Path,
    delta_filename: str,
    target_platform: str,
    delta_cmd: Optional[list[str]] = XDELTA_CMD,
    logger: Optional[logging.Logger] = 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
        with open(source_path, "w") as f:
            f.write(from_layers[i].name)

    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 inclusion in the cache by image ref and hash, or by OCI image filename.

ImageCache 

ImageCache(cache_dir: Path, delta_cache_dirname: Optional[str] = DELTA_CACHE_DIRNAME, extract_cache_dirname: Optional[str] = EXTRACT_CACHE_DIRNAME, image_file_name: Optional[str] = IMAGE_FILE_NAME, save_cache_dirname: Optional[str] = SAVE_CACHE_DIRNAME, logger: Optional[Logger] = 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 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.

  • delta_cache_dirname

    (Optional[str], default: DELTA_CACHE_DIRNAME ) –

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

  • extract_cache_dirname

    (Optional[str], default: EXTRACT_CACHE_DIRNAME ) –

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

  • image_file_name

    (Optional[str], default: IMAGE_FILE_NAME ) –

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

  • save_cache_dirname

    (Optional[str], default: SAVE_CACHE_DIRNAME ) –

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

  • logger

    (Optional[Logger], default: getLogger(__name__) ) –

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

Methods:

  • 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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    cache_dir: Path,
    delta_cache_dirname: Optional[str] = DELTA_CACHE_DIRNAME,
    extract_cache_dirname: Optional[str] = EXTRACT_CACHE_DIRNAME,
    image_file_name: Optional[str] = IMAGE_FILE_NAME,
    save_cache_dirname: Optional[str] = SAVE_CACHE_DIRNAME,
    logger: Optional[logging.Logger] = 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 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

    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()
    }

_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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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."
        )

delta 

delta(from_image: dict[str, str], to_image: dict[str, str], target_platform: 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.

  • target_platform
    (str) –

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

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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def delta(
    self, from_image: dict[str, str], to_image: dict[str, str], target_platform: 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.
    :param target_platform: The platform to target, if the image contains multiple, as os/[architecture]/[variant].
    :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,
            target_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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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]
        image_ref = image_manifest.get("annotations", {}).get(
            "io.containerd.image.name", None
        )
        image_hash = image_manifest.get("digest", None)
        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 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
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.logger.debug(f"Image file {save_image_file} saved.")
        self.save_cache[image_hash] = save_image_file
        return save_image_file

mender_docker_lifecycle_helper.utils.mender_server

Functions:

  • call_mender_host_api –

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

call_mender_host_api 

call_mender_host_api(context: LifecycleHelperContext, mender_endpoint: str, request_args: dict) -> 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.

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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def call_mender_host_api(
    context: LifecycleHelperContext,
    mender_endpoint: str,
    request_args: dict,
) -> 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.
    :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

    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:
        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}"
        )
        r.raise_for_status()
    else:
        return r