Convert Openshift `DeploymentConfig` to Kubernetes `Deployment`

oc_to_k8s.py

"""
Convert Openshift `DeploymentConfig` to Kubernetes `Deployment`.

requirements:
    pip install ruamel.yaml
"""


import argparse
import json
import logging.config
import os
import shutil
from copy import deepcopy
from pathlib import Path
from typing import Any

from ruamel.yaml import YAML

BASE_DIR = Path.cwd()
OC_DIR = BASE_DIR / "openshift"
DEFAULT_OC_TEMPLATE = BASE_DIR / "openshift" / "template.yaml"

yaml = YAML()
yaml.default_flow_style = False
yaml.width = 120
yaml.preserve_quotes = True

logger = logging.getLogger(__name__)


def read_config(filename: str | os.PathLike, **load_kwargs) -> dict[str, Any]:
    """
    Load configuration settings from a YAML or JSON file.

    :param filename: Path to the configuration file.
    :param load_kwargs: Additional keyword arguments for the load function.
    :return: Configuration settings as a dictionary.
    """
    extension = Path(filename).suffix.lower()
    logger.debug(f"parsing config file: [{filename}]")

    if extension in (".yaml", ".yml"):
        load_func = yaml.load
    elif extension in (".json", ".jsn"):
        load_func = json.load
    else:
        raise ValueError(
            f"Unsupported file extension '{extension}'. "
            f"Only .yaml, .yml, .json, and .jsn are supported."
        )

    with open(filename) as file:
        return load_func(file, **load_kwargs)


def save_to_yaml(data: dict | list, file: str | os.PathLike, **dump_kwargs):
    """
    Save given data to a YAML file, handling multi-line strings gracefully.

    :param data: dict | list
        A data that should be saved to YAML file.
    :param file: str | os.PathLike
        Filename as a string or a pathlib object.
    :param dump_kwargs: named arguments
        Named arguments that will be passed to `ruamel.yaml.YAML.dump()` function.
    :return:
    """
    file_name = str(file)
    logger.debug(f"saving data to YAML file: [{file_name}]")

    # Determine how to handle the file parameter and save the YAML data accordingly
    if isinstance(file, str | Path):
        # Ensure the directory exists
        directory = Path(file).parent
        directory.mkdir(parents=True, exist_ok=True)

        with open(file, "w") as f:
            yaml.dump(data, f, **dump_kwargs)
    else:
        raise ValueError(
            "The file argument must be a filename or a pathlib.Path object"
        )


def setup_logging(config_file: str | os.PathLike | None = None):
    """
    Configure logging based on a configuration file.

    :param config_file: Path to the configuration file.
    """
    if config_file is None:
        logging_conf_str = """
            version: 1
            disable_existing_loggers: false
            formatters:
              simple:
                format: '%(asctime)s|%(module)s|%(levelname)s|%(message)s'
                datefmt: '%Y-%m-%dT%H:%M:%S%z'
            handlers:
              console:
                class: logging.StreamHandler
                formatter: simple
                stream: ext://sys.stdout
            loggers:
              root:
                level: INFO
                handlers:
                - console
        """
        config = yaml.load(logging_conf_str)
    else:
        config = read_config(config_file)
    logging.config.dictConfig(config)


def convert_oc_to_k8s(template: dict[str, Any]) -> dict[str, Any]:
    """
    Convert Openshift `DeploymentConfig` to Kubernetes `Deployment`.

    1. skip object if its kind != "DeploymentConfig"
    2. replace `apiVersion: apps.openshift.io/v1` -> `apiVersion: apps/v1`
    3. replace `kind: DeploymentConfig` -> `kind: Deployment`
    4. replace `spec.selectors` from `selector: name: ...` -> `selector: matchLabels: name: ...`
    5. ensure the `spec.template.spec.containers.image` section is defined for each container
    6. delete the `spec.triggers`, `spec.strategy`, and `spec.test` sections

    :param template: Openshift configuration
    :return: Kubernetes configuration
    """
    logger.info("converting Openshift template: `DeploymentConfig` -> `Deployment`")
    data = deepcopy(template)
    for idx, obj in enumerate(data["objects"]):
        # 1. skip object if its kind != "DeploymentConfig"
        if obj["kind"] != "DeploymentConfig":
            logger.debug(f"leaving object with `kind: {obj['kind']}` unchanged")
            continue
        # 2. replace `apiVersion: apps.openshift.io/v1` -> `apiVersion: apps/v1`
        logger.debug("setting `apiVersion: apps/v1` ")
        obj["apiVersion"] = "apps/v1"
        # 3. replace `kind: DeploymentConfig` -> `kind: Deployment`
        logger.debug("setting `kind: Deployment` ")
        obj["kind"] = "Deployment"
        # 4. replace `spec.selectors` from `selector: name: ...` -> `selector: matchLabels: name: ...`
        logger.debug("adding `matchLabels` to `spec.selector` ")
        if "selector" in obj.get("spec"):
            selector = deepcopy(obj["spec"]["selector"])
            obj["spec"]["selector"].clear()
            obj["spec"]["selector"]["matchLabels"] = selector
        # 5. ensure the `spec.template.spec.containers.image` section is defined for each container
        for i, container in enumerate(obj["spec"]["template"]["spec"]["containers"]):
            logger.debug(f"ensuring container: `{container['name']}` has `image` key")
            if "image" not in container:
                raise ValueError(
                    f"""No `image` key found in `container` with name: {container["name"]}."""
                    f""" Path in `DeploymentConfig`: spec.template.spec.containers[{i}]"""
                )
        # 6. delete the `spec.triggers`, `spec.strategy`, and `spec.test` sections
        logger.debug(
            "removing `spec.triggers`, `spec.strategy` and `spec.test` sections"
        )
        if "triggers" in obj.get("spec"):
            del obj["spec"]["triggers"]
        if "strategy" in obj.get("spec"):
            del obj["spec"]["strategy"]
        if "test" in obj.get("spec"):
            del obj["spec"]["test"]
    return data


def reorder_oc_template(
    template: dict[str, Any],
    kind_first: bool = True,
    params_to_top: bool = False,
) -> dict[str, Any]:
    logger.info("reordering Openshift template")
    data = deepcopy(template)
    if kind_first:
        for idx, obj in enumerate(data["objects"]):
            if "kind" in obj:
                data["objects"][idx] = {"kind": obj["kind"]} | obj
        if "kind" in data:
            data = {"kind": data["kind"]} | data
    if params_to_top and "parameters" in data:
        data = {"parameters": data["parameters"]} | data
    return data


def parse_args():
    """Parse command line arguments."""
    logger.debug("Parsing command line arguments")
    examples = """
    python oc_to_k8s.py -i openshift/template.yaml -o openshift/template_new.yaml

will convert `openshift/template.yaml` file into `openshift/template_new.yaml`.
"""
    desc = (
        f"Converts `DeploymentConfig` section in the OpenShift `template.yaml` into `Deployment`:\n"
        f"\n- replace `apiVersion: apps.openshift.io/v1` -> `apiVersion: apps/v1`"
        f"\n- replace `kind: DeploymentConfig` -> `kind: Deployment`"
        f"\n- replace `spec.selectors` from `selector: name: ...` to `selector: matchLabels: name: ...`"
        f"\n- make sure that the `spec.template.spec.containers.image` section is defined for each container"
        f"\n- delete the `spec.triggers`, `spec.strategy`, and `spec.test` sections"
        f"\n- reorder keys - "
        f"\n\nUsage examples:\n{examples}"
    )
    parser = argparse.ArgumentParser(
        description=desc, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "-i",
        "--input-file",
        help=f"Input OpenShift template YAML file (default: [{DEFAULT_OC_TEMPLATE}])",
        default=DEFAULT_OC_TEMPLATE,
    )
    parser.add_argument(
        "-o",
        "--output-file",
        help=f"Output OpenShift template YAML file (default: [{DEFAULT_OC_TEMPLATE}])",
        default=DEFAULT_OC_TEMPLATE,
    )
    parser.add_argument(
        "-p",
        "--params-to-top",
        action="store_true",
        help=f"Reorder `parameters` key to the top of resulting YAML file (default: [{False}])",
        default=False,
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="count",
        default=0,
        help="Increase verbosity level. Specify once for verbose output, "
        "twice for more detailed output.",
    )
    return parser.parse_args()


def main():
    """Define main function to set up and execute script functionalities."""
    setup_logging()
    args = parse_args()

    if args.verbose > 0:
        # set logging level DEBUG if `--verbose` flag was specified
        logger.setLevel(logging.DEBUG)

    input_file = Path(args.input_file).resolve()
    output_file = Path(args.output_file).resolve()

    if args.input_file == args.output_file:
        bkp_file = output_file.with_stem(
            f"{output_file.stem}.DeploymentConfig{output_file.suffix}"
        ).with_suffix(".bkp")
        logger.info(f"creating backup file: [{bkp_file}]")
        shutil.copy(input_file, bkp_file)

    oc_template = read_config(input_file)
    new_template = reorder_oc_template(
        template=convert_oc_to_k8s(oc_template),
        kind_first=True,
        params_to_top=args.params_to_top,
    )
    save_to_yaml(new_template, output_file)


if __name__ == "__main__":
    main()