Update cratedb over all namespace (label selector)

gc_image_updater.py

#!/usr/bin/env python3

# https://pip.wtf
def pip_wtf(command):
    import os, os.path, sys
    t = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".pip_wtf." + os.path.basename(__file__))
    sys.path = [p for p in sys.path if "-packages" not in p] + [t]
    os.environ["PATH"] += os.pathsep + t + os.path.sep + "bin"
    os.environ["PYTHONPATH"] = os.pathsep.join(sys.path)
    if os.path.exists(t): return
    os.system(" ".join([sys.executable, "-m", "pip", "install", "-t", t, command]))

# Now you just call it to install your packages:
#   pip_wtf('the rest of the pip install command here')
# Here are some examples for different platforms:

import sys
pip_wtf('kubernetes icecream halo datetime semver subprocess.run logging colorlog')

from kubernetes import client, config, utils
from kubernetes.client.rest import ApiException
from icecream import ic
from datetime import datetime as dt
import os
import subprocess
import json
import semver
import logging
import colorlog
from colorlog.escape_codes import escape_codes


# Optional environment settings
CONTEXT = os.getenv('KUBE_CONTEXT', 'aks1-eastus-dev')
IMAGE_NAME = os.getenv('IMAGE_NAME', 'crate/grand-central')
NS_LABEL_SELECTOR = os.getenv('LABEL_SELECTOR', 'app.kubernetes.io/part-of=cratedb')
REGISTRY_NAME = os.getenv('REGISTRY_NAME', 'cr8cloud')

# Always use the latest available tag
LATEST_ONLY = os.getenv('LATEST_ONLY', 'false')

def setup_logging(level=logging.INFO, debug: bool = False, width: int = 16):
    reset = escape_codes["reset"]
    log_format = f"%(asctime)-15s [%(name)-{width}s] %(log_color)s%(levelname)-8s:{reset} %(message)s"

    handler = colorlog.StreamHandler()
    handler.setFormatter(colorlog.ColoredFormatter(log_format))

    logging.basicConfig(format=log_format, level=level, handlers=[handler])

# Get the latest tag from the registry via Azure CLI
def get_latest_tag(registry_name, image_name):
    # Call the Azure CLI to get the list of tags
    result = subprocess.check_output(
        [
            "az", "acr", "repository", "show-tags",
            "--name", registry_name,
            "--repository", image_name,
            "--orderby", "time_desc",
            "--output", "json",
        ],
        text=True,
    )

    # Parse the output as JSON
    tags = json.loads(result)

    # Filter out the tags that are not valid semver versions
    semver_tags = [tag for tag in tags if is_semver(tag)]

    # Get the latest semver tag
    if not semver_tags or LATEST_ONLY == 'true':
        latest_semver_tag = tags[0]
    else:
        latest_semver_tag = semver_tags[0]

    return latest_semver_tag

# Check wether a tag is a valid semver version
def is_semver(tag):
    try:
        semver.parse_version_info(tag)
        return True
    except ValueError:
        return False

# Prompt for icecream output
def ts():
   return '%s |> ' % dt.now().strftime('%Y-%m-%d %H:%M:%S')

setup_logging(level=logging.INFO, debug=True)
logger = logging.getLogger("gc_image_updater")
logger.info("//Starting")
logger.info("acr %s Registry: %s", REGISTRY_NAME, IMAGE_NAME)
#ic.configureOutput(prefix=ts)
#ic(REGISTRY_NAME, IMAGE_NAME)

logger.info("Looking up latest tag from ACR via Azure CLI for %s", IMAGE_NAME)
latest_tag = get_latest_tag(REGISTRY_NAME, IMAGE_NAME)
actual_latest_tag = IMAGE_NAME + ":" + latest_tag
logger.info("Latest tag found %s", actual_latest_tag)
if LATEST_ONLY == 'true':
    logger.WARNING("LATEST_ONLY is set to true, so we will use the latest tag")

# switch to a specific k8s context provided by an environment variable
config.load_kube_config(context=CONTEXT)
crd = client.CustomObjectsApi()
core_api = client.CoreV1Api()

API_GROUP = "cloud.crate.io"
RESOURCE_CRATEDB = "cratedbs"

# Print the k8s context we are working on
logger.info("k8s context: %s", CONTEXT)

# List all namespaces with the label corresponding NS_LABEL_SELECTOR
namespaces = core_api.list_namespace(label_selector=NS_LABEL_SELECTOR)

# Loop over all namespace with the label corresponding NS_LABEL_SELECTOR
for namespace in namespaces.items:
    logger.info("// namespace: %s", namespace.metadata.name)
    cratedbs = crd.list_namespaced_custom_object(API_GROUP, "v1", namespace.metadata.name, RESOURCE_CRATEDB)
    for cratedb in cratedbs["items"]:
        logger.info("cratedb: %s %s", cratedb["metadata"]["name"], cratedb["spec"]["cluster"]["name"])
        name = cratedb["metadata"]["name"]
        try:
           # Get the existing custom object
           custom_object = crd.get_namespaced_custom_object(API_GROUP, "v1", namespace.metadata.name, RESOURCE_CRATEDB, name)
           logger.info("cratedb: %s", custom_object["spec"]["cluster"]["version"])

           # Update the image in the custom object
           if "grand-central" in custom_object["spec"]:
               old_image = custom_object["spec"]["grand-central"]["image"]
               logger.info("Old image: %s", old_image)
               custom_object["spec"]["grand-central"]["image"] = actual_latest_tag
               # Apply the updated custom object
               # //TODO: this needs to be commented out, only for testing
               #crd.patch_namespaced_custom_object(API_GROUP, "v1", namespace.metadata.name, RESOURCE_CRATEDB, cratedb)
           else:
               logger.warning("grand-central probably not configured.")

        except ApiException as e:
            if e.status == 404:
                logger.error("cratedb not found!")
            else:
                raise