Image Replicator¶

Container image replication utilities.

ECRImageReplicator ¶

Bases: LoggingMixin

put_image ¶

put_image(
    source_image,
    destination_repository,
    destination_image_tags=None,
)

Put image manifest and tags for an image.

This is the final step in copying an ECR image. It must be done once all the layers of the image have been uploaded.

If the put_image request fails with a 'LayersNotFoundException', this will attempt to upload the missing layers and retry the put_image request.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Source image that has been copied.	required
`destination_repository`	`ECRRepository`	Destination repository.	required
`destination_image_tags`	`Optional[List[str]]`	Destination image tags. Optional. If not provided, source image's tags are added.	`None`

Returns:

Type	Description
`ECRImage`	Destination ECR Image.

Source code in src/aibs_informatics_aws_utils/ecr/image_replicator.py

def put_image(
    self,
    source_image: ECRImage,
    destination_repository: ECRRepository,
    destination_image_tags: Optional[List[str]] = None,
) -> ECRImage:
    """Put image manifest and tags for an image.

    This is the final step in copying an ECR image. It must be done
    once all the layers of the image have been uploaded.

    If the put_image request fails with a 'LayersNotFoundException',
    this will attempt to upload the missing layers and retry
    the put_image request.

    Args:
        source_image: Source image that has been copied.
        destination_repository: Destination repository.
        destination_image_tags: Destination image tags. Optional.
            If not provided, source image's tags are added.

    Returns:
        Destination ECR Image.
    """
    tags = (
        source_image.image_tags if destination_image_tags is None else destination_image_tags
    )

    self.logger.info(f"putting image to {destination_repository.uri} with {tags} tags")
    dest_image = ECRImage(
        account_id=destination_repository.account_id,
        region=destination_repository.region,
        repository_name=destination_repository.repository_name,
        image_digest=source_image.image_digest,
        image_manifest=source_image.image_manifest,
        client=destination_repository.client,
    )
    dest_image.client = destination_repository.client
    try:
        dest_image.put_image(None)
        if tags:
            dest_image.add_image_tags(*tags)

    except ClientError as e:
        # IF we have a LayersNotFoundException, then we will retry to upload layers
        if get_client_error_code(e) != "LayersNotFoundException":
            self.log_stacktrace(
                f"Error putting image to {destination_repository.uri} with {tags}",
                e,
            )
            raise e
        self.logger.warning(
            f"Received 'LayerNotFoundException' while putting image. "
            f"Resolving missing layers of {source_image} specified in {e}"
        )
        missing_layers = self._get_missing_layers(
            client=source_image.client,
            repository_name=source_image.repository_name,
            put_image_error=e,
        )
        self.logger.info(f"Identified {len(missing_layers)} missing layers. Uploading layers")
        self._upload_layers(
            source_repository=source_image.get_repository(),
            destination_repository=destination_repository,
            layers=missing_layers,
            check_if_exists=False,
        )

        self.logger.info("uploaded all missing layers, retrying put_image request")
        self.put_image(source_image, destination_repository, destination_image_tags)
    return dest_image

replicate ¶

replicate(
    source_image,
    destination_repository,
    destination_image_tags=None,
)

Copies the source image into the destination repository.

This allows the user to facilitate replication:

between AWS accounts
to a new repository
with new tags

Note

This has exactly the same effect as docker pull; docker tag; docker push, but is done with the AWS ECR SDK so we can run this in a lambda when our stack updates.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Configuration of image to pull.	required
`destination_repository`	`ECRRepository`	Repository to push to.	required
`destination_image_tags`	`Optional[List[str]]`	Optional list of tags to apply to the destination image.	`None`

Returns:

Type	Description
`ECRImage`	Configuration of destination image pushed.

Source code in src/aibs_informatics_aws_utils/ecr/image_replicator.py

def replicate(
    self,
    source_image: ECRImage,
    destination_repository: ECRRepository,
    destination_image_tags: Optional[List[str]] = None,
) -> ECRImage:
    """Copies the source image into the destination repository.

    This allows the user to facilitate replication:

    - **between AWS accounts**
    - **to a new repository**
    - **with new tags**

    Note:
        This has exactly the same effect as `docker pull; docker tag; docker push`,
        but is done with the AWS ECR SDK so we can run this in a lambda when our
        stack updates.

    Args:
        source_image: Configuration of image to pull.
        destination_repository: Repository to push to.
        destination_image_tags: Optional list of tags to apply to the destination image.

    Returns:
        Configuration of destination image pushed.
    """
    self.log.info(
        f"Starting Image Replication "
        f"[source={source_image.uri}, destination={destination_repository.uri}] "
        f"with tags={destination_image_tags}"
    )

    self.upload_layers(
        source_image=source_image, destination_repository=destination_repository
    )

    self.logger.info(f"Putting image to {destination_repository.uri}")

    self.put_image(
        source_image=source_image,
        destination_repository=destination_repository,
        destination_image_tags=destination_image_tags,
    )
    self.logger.info(f"Completed putting image to {destination_repository.uri}.")

    return ECRImage(
        account_id=destination_repository.account_id,
        region=destination_repository.region,
        repository_name=destination_repository.repository_name,
        image_digest=source_image.image_digest,
        client=destination_repository.client,
    )

upload_layers ¶

upload_layers(source_image, destination_repository)

Upload image layers of the source image to the destination repository.

Parameters:

Name	Type	Description	Default
`source_image`	`ECRImage`	Source image that has been copied.	required
`destination_repository`	`ECRRepository`	Destination repository.	required

Source code in src/aibs_informatics_aws_utils/ecr/image_replicator.py

def upload_layers(self, source_image: ECRImage, destination_repository: ECRRepository):
    """Upload image layers of the source image to the destination repository.

    Args:
        source_image: Source image that has been copied.
        destination_repository: Destination repository.
    """
    self.logger.info(
        f"Uploading layers from {source_image.uri} to {destination_repository.uri}."
    )

    layers = source_image.get_image_layers()
    config_layer = source_image.get_image_config_layer()

    all_layers = layers + [config_layer]

    self.logger.info(
        f"Source image {source_image.uri} has {len(layers)} layers and "
        f"1 config layer totalling {len(all_layers)} layers"
    )

    self._upload_layers(
        source_repository=source_image.get_repository(),
        destination_repository=destination_repository,
        layers=all_layers,
        check_if_exists=True,
    )