dcqc/main.py

#!/usr/bin/env python3
import argparse
import sys
from pathlib import Path

import yaml


class IniFile:
    PREAMBLE = "### Generated by dcqc - do not edit ###"
    def __init__(self):
        self._data = {}

    def __getitem__(self, key: str):
        return self._data[key]

    def __setitem__(self, key: str, value):
        self._data[key] = value

    def write(self, f):
        f.write(self.PREAMBLE)
        f.write("\n\n")
        for section, section_data in self._data.items():
            f.write(f"[{section}]\n")
            for key, value in section_data.items():
                if isinstance(value, list):
                    for item in value:
                        f.write(f"{key}={item}\n")
                else:
                    f.write(f"{key}={value}\n")
            f.write("\n")

def abort(msg: str):
    print(msg, file=sys.stderr)
    sys.exit(1)

def enforce_list(key: str, dict_: dict):
    if not isinstance(dict_[key], list):
        abort(f"{key} is not a list. Offending dict: {dict_}")


def write_file(ini_file: IniFile, path: Path):
    with open(path, "w", encoding="utf-8") as f:
        ini_file.write(f)

def write_build_unit(args, yaml_dict: dict, service_name: str):
    out_file = args.output_dir / f"{service_name}.build"
    print(f'Generating build "{service_name}" ({out_file})')
    build = yaml_dict["services"][service_name]["build"]

    unit_file = IniFile()
    unit_file["Unit"] = {}
    unit_file["Unit"]["Description"] = f"{service_name.capitalize()} build"

    unit_file["Build"] = {}
    unit_file["Build"]["ImageTag"] = f"localhost/{service_name}"
    if isinstance(build, str):
        unit_file["Build"]["SetWorkingDirectory"] = build
    elif isinstance(build, dict):
        if "context" in build:
            unit_file["Build"]["SetWorkingDirectory"] = build["context"]
        if "dockerfile" in build:
            unit_file["Build"]["File"] = build["dockerfile"]

    write_file(unit_file, out_file)


def write_network_units(args, yaml_dict):
    for network_name in yaml_dict.get("networks", []):
        out_file = args.output_dir / f"{network_name}.network"
        print(f'Generating network "{network_name}" ({out_file})')
        network = yaml_dict["networks"].get(network_name)
        if network is None:
            network = {}

        unit_file = IniFile()
        unit_file["Unit"] = {}
        unit_file["Unit"]["Description"] = f"{network_name.capitalize()} network"

        unit_file["Network"] = {}
        unit_file["Network"]["NetworkName"] = network_name
        unit_file["Network"]["Internal"] = str(network.get("internal", True)).lower()
        if "enable_ipv6" in network:
            unit_file["Network"]["IPv6"] = str(network["enable_ipv6"]).lower()

        unit_file["Service"] = {}
        unit_file["Service"]["Restart"] = "on-failure"

        unit_file["Install"] = {}
        unit_file["Install"]["WantedBy"] = "default.target"

        write_file(unit_file, out_file)


def write_service_units(args, yaml_dict):
    for service_name in yaml_dict["services"]:
        out_file = args.output_dir / f"{service_name}.container"
        print(f'Generating container "{service_name}" ({out_file})')
        service = yaml_dict["services"][service_name]

        unit_file = IniFile()
        unit_file["Unit"] = {}
        unit_file["Unit"]["Description"] = f"{service_name.capitalize()} container"
        if "depends_on" in service:
            enforce_list("depends_on", service)
            for dependency in service["depends_on"]:
                unit_file["Unit"].setdefault("Requires", []).append(f"{dependency}.service")
                unit_file["Unit"].setdefault("After", []).append(f"{dependency}.service")

        unit_file["Container"] = {}
        if "container_name" in service:
            unit_file["Container"]["ContainerName"] = service["container_name"]
        if "image" in service:
            unit_file["Container"]["Image"] = service["image"]
            unit_file["Container"]["AutoUpdate"] = "registry"
        elif "build" in service:
            write_build_unit(args, yaml_dict, service_name)
            unit_file["Container"]["Image"] = f"{service_name}.build"
            unit_file["Container"]["AutoUpdate"] = "local"
        if "ports" in service:
            enforce_list("ports", service)
            unit_file["Container"]["PublishPort"] = service["ports"]
        if "command" in service:
            enforce_list("command", service)
            command = service["command"]
            cmd = command[0]
            if len(command) > 1:
                cmd += " \\\n"
            for i, cmd_part in enumerate(command[1:]):
                cmd += f"  {cmd_part}"
                if i < len(command) - 2:
                    cmd += " \\\n"
            unit_file["Container"]["Exec"] = cmd
        if "labels" in service:
            enforce_list("labels", service)
            unit_file["Container"]["Label"] = service["labels"]
        if "environment" in service:
            if isinstance(service["environment"], dict):
                env_vars = []
                for key, value in service["environment"].items():
                    env_vars.append(f"{key}={value}")
                service["environment"] = env_vars
            enforce_list("environment", service)
            unit_file["Container"]["Environment"] = service["environment"]
        if "env_file" in service:
            unit_file["Container"]["EnvironmentFile"] = service["env_file"]
        if "volumes" in service:
            enforce_list("volumes", service)
            unit_file["Container"]["Volume"] = service["volumes"]
        if "devices" in service:
            unit_file["Container"]["AddDevice"] = service["devices"]
        if "userns_mode" in service:
            unit_file["Container"]["UserNS"] = service["userns_mode"]
        if "group_add" in service:
            enforce_list("group_add", service)
            unit_file["Container"]["GroupAdd"] = service["group_add"]


        unit_file["Service"] = {}
        unit_file["Service"]["Restart"] = "on-failure"

        unit_file["Install"] = {}
        unit_file["Install"]["WantedBy"] = "default.target"

        write_file(unit_file, out_file)


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("compose_file")
    parser.add_argument("-o", "--output-dir", default="/etc/containers/systemd", type=Path)
    return parser.parse_args()


def main():
    args = parse_args()
    with open(args.compose_file, encoding="utf-8") as f:
        yaml_dict = yaml.safe_load(f)

    write_network_units(args, yaml_dict)
    write_service_units(args, yaml_dict)

    print('\n==> Remember to run "systemctl daemon-reload"')



if __name__ == '__main__':
    main()