import os
from typing import List, Dict, Any
import shutil
import yaml
from jinja2 import Environment, FileSystemLoader, select_autoescape, StrictUndefined
TEMPLATE_PATH = os.path.join(os.path.dirname(__file__), "templates")
COPY_PATH = os.path.join(os.path.dirname(__file__), "non_templates")
[docs]
class RelEnvironment(Environment):
"""Override join_path() to enable relative template paths."""
[docs]
def join_path(self, template, parent):
return os.path.normpath(os.path.join(os.path.dirname(parent), template))
[docs]
def render_config(
helm_config_path: str,
output_path: str,
slug: str,
environment: str,
instance_override_compose_paths: List[str],
) -> None:
# get helm data
configmaps = get_helm_data(helm_config_path, file_type="configmap")
deployments = get_helm_data(helm_config_path, file_type="deployment")
env = get_helm_data(helm_config_path, file_type="deployment", index=True)
ingress = get_helm_data(helm_config_path, file_type="ingress", index=True)
# extract envs
envs = extract_env_vars(env)
service_configs = get_service_configs(deployments, ingress)
# merge envs and params
merged_params = {
**envs,
"slug": slug,
"environment": environment,
"service_configs": service_configs,
}
# prepare the configs
mapping = prepare_configs(configmaps)
# create folder with name of slug at path
destination = prepare_directory(
output_path,
slug,
environment=environment,
)
# create necessary configs and store at appropriate path
create_configurations(destination, mapping)
# fill remaining templates and store at appropriate path
render_templates(destination, instance_override_compose_paths, merged_params)
# create copies of files, which are not to be considered templates
copy_non_template_files(destination)
[docs]
def get_helm_data(helm_config_path: str, file_type: str, index: bool = False) -> Any:
if not os.path.isdir(helm_config_path):
raise ValueError(f"{helm_config_path} is not a directory")
rendered_templates = os.listdir(helm_config_path)
files = []
for template_dir in rendered_templates:
template_path = os.path.join(helm_config_path, template_dir)
for f in os.listdir(template_path):
if file_type in f and os.path.isfile(
template_file := os.path.join(template_path, f)
):
files.append(template_file)
if index:
for i, f in enumerate(files):
if "renderer" in f:
return files[i]
return files
[docs]
def prepare_configs(configmaps: List[str]) -> Dict[str, str]:
file_config_mapping = {}
for configmap in configmaps:
config = yaml_open(configmap)
config = config["data"]
if config:
for file_name in config.keys():
config_content = config[file_name]
file_config_mapping[file_name] = config_content
return file_config_mapping
[docs]
def prepare_directory(output_path: str, slug: str, environment: str) -> str:
destination = os.path.join(output_path, f"{slug}-{environment}")
try:
os.makedirs(destination)
except FileExistsError:
pass
return destination
[docs]
def create_configurations(destination: str, mapping: Dict[str, str]) -> None:
destination = os.path.join(destination, "config")
try:
os.mkdir(destination)
except FileExistsError:
pass
for filename, config in mapping.items():
config_path = os.path.join(destination, os.path.basename(filename))
with open(config_path, "w", encoding="utf-8") as cfg:
cfg.write(config)
[docs]
def render_templates(
destination: str, instance_override_compose_paths: List[str], params: Dict[str, Any]
) -> None:
template_folders = [TEMPLATE_PATH, "/"]
instance_override_compose_paths_absolute = []
for extra_path in instance_override_compose_paths:
# convert to abspaths for clarity
abs_path = os.path.abspath(os.path.expanduser(os.path.expandvars(extra_path)))
folder = abs_path.rsplit(os.sep, 1)[0]
template_folders.append(folder)
instance_override_compose_paths_absolute.append(abs_path)
jinja_env = RelEnvironment(
loader=FileSystemLoader(template_folders),
autoescape=select_autoescape(),
undefined=StrictUndefined,
)
templates = os.listdir(TEMPLATE_PATH) + instance_override_compose_paths_absolute
for template in templates:
tpl = jinja_env.get_template(template)
rendered_template = tpl.render(**params)
template_destination = (
os.path.join(destination, os.path.basename(template))
if "docker" in template
else os.path.join(destination, "config", os.path.basename(template))
)
with open(template_destination, "w", encoding="utf-8") as tpl_file:
tpl_file.write(rendered_template)
[docs]
def copy_non_template_files(destination: str) -> None:
if os.path.exists(COPY_PATH):
files_to_copy = os.listdir(COPY_PATH)
for file_to_copy in files_to_copy:
file_destination = os.path.join(destination, "config", file_to_copy)
shutil.copyfile(os.path.join(COPY_PATH, file_to_copy), file_destination)
[docs]
def yaml_open(path: str) -> Any:
with open(path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
[docs]
def get_service_configs(
deployments: List[str], ingress: str
) -> Dict[str, Dict[str, str]]:
service_configs: Dict[str, Dict[str, str]] = {}
# extract some properties from rendered k8s files
for d in deployments:
yml = yaml_open(d)
service = yml["spec"]["template"]["spec"]["containers"][0]["name"]
service_configs[service] = {}
# extract image + version
service_configs[service]["image"] = yml["spec"]["template"]["spec"][
"containers"
][0]["image"]
# extract number of replicas
service_configs[service]["replicas"] = yml["spec"]["replicas"]
# extract ingress hostname
yml = yaml_open(ingress)
service_configs["host"] = yml["spec"]["rules"][0]["host"]
return service_configs