Compare Google Container Registry and Artefact Registry repos. This is useful when you are migrating from GCR to AR (since GCP is deprecating GCR in May 2024). This script helps you to be sure that all images are copied to AR.

compare_gcr_with_ar.py

import json
import os
import subprocess
from pprint import pprint
from typing import List


PROJECT_ID = os.getenv("PROJECT_ID")
GCR_URL = os.getenv("GCR_URL")  # eu.gcr.io
AR_REGION = os.getenv("AR_REGION")  # e.g. europe
AR_REPO = os.getenv("AR_REPO")


class ImageInfo:
    def __init__(self, name: str, digest: str, tags: List[str] | str):
        self.name = name.split("/")[-1]
        self.digest = digest
        if tags:
            if isinstance(tags, str):
                self.tags = set([t.strip() for t in tags.split(",")])
            else:
                self.tags = set([t.strip() for t in tags])
        else:
            self.tags = None

    def __eq__(self, other):
        if isinstance(other, ImageInfo):
            return self.__hash__ == other.__hash__()
        return False

    def __hash__(self):
        return hash(self.__str__())

    def __str__(self):
        return f"{self.name}:{self.digest}:{self.tags}"


def run_cmd_command(cmd: str) -> str:
    print(f"Running command: {cmd}")
    result = subprocess.run(cmd.split(), stdout=subprocess.PIPE)
    return result.stdout.decode('utf-8')


def get_gcr_image_names():
    list_images_gcr_cmd = f"gcloud container images list --repository={GCR_URL}/{PROJECT_ID}"
    result = run_cmd_command(list_images_gcr_cmd)
    result = result.split("\n")[1:]
    return [s for s in result if s]


def get_gcr_image_infos() -> List[ImageInfo]:
    gcr_image_names = get_gcr_image_names()
    result = []
    for image_name in gcr_image_names:
        list_image_info_cmd = f"gcloud container images list-tags {image_name} --format=\"json\""
        stdout = run_cmd_command(list_image_info_cmd)
        info = json.loads(stdout)
        result.extend([ImageInfo(name=image_name, digest=i['digest'], tags=i['tags']) for i in info])
    return result


def get_ar_image_infos() -> List[ImageInfo]:
    cmd = f"gcloud artifacts docker images list {AR_REGION}-docker.pkg.dev/{PROJECT_ID}/{AR_REPO} --include-tags --format=\"json\""
    stdout = run_cmd_command(cmd)
    info = json.loads(stdout)
    result = []
    for i in info:
        result.append(ImageInfo(name=i['package'], digest=i['version'], tags=i['tags']))
    return result


if __name__ == '__main__':
    gcr_images = get_gcr_image_infos()
    ar_images = get_ar_image_infos()

    sorted_gcr_images = set(sorted(gcr_images, key=lambda image: (image.name, image.digest, image.tags)))
    sorted_ar_images = set(sorted(ar_images, key=lambda image: (image.name, image.digest, image.tags)))

    stringified_gcr_set = set([image.__str__() for image in sorted_gcr_images])
    stringified_ar_set = set([image.__str__() for image in sorted_ar_images])

    print(f"Missing in Artifact Registry ({AR_REGION}-docker.pkg.dev/{PROJECT_ID}/{AR_REPO}) :")
    pprint(stringified_gcr_set - stringified_ar_set)

    print(f"Missing in Google Cloud Registry ({GCR_URL}/{PROJECT_ID}):")
    pprint(stringified_ar_set - stringified_gcr_set)