infra/metal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Render all template using python and jinja

Browse Source
This commit is contained in:
Dreaded_X
2025-11-08 06:14:30 +01:00
parent 3290a76193
commit d30b080098
5 changed files with 147 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.editorconfig
View File

@@ -7,6 +7,6 @@ indent_style = tab
indent_style = space
indent_size = 4
[{*.py,tools/merge}]
[{*.py,tools/render}]
indent_style = space
indent_size = 4

25
templates/boot.ipxe
View File

@@ -5,21 +5,18 @@ dhcp
echo Starting ${serial}
:start
# Is a known serial is set, execute that
# If an unknown serial is set, exit
# If no serial is set, ask the user
goto node_${serial} || shell
goto node_${serial} || exit
# Default behavior (non install mode) is to exit iPXE script
{{ range datasource "nodes" }}
{{- if .install }}
# {{ .filename }}
:node_{{ .serial }}
{{- $ipArg := printf "ip=%s::%s:%s:%s:%s::%s:%s:%s" .ip .gateway .netmask .hostname .interface (index .dns 0) (index .dns 1) .ntp }}
{{- $kernelArgs := printf "%s %s %s" $ipArg (join .kernelArgs " ") (join .extraKernelArgs " ") }}
{% for node in nodes %}
{%- if node.install -%}
# {{ node.filename }}
:node_{{ node.serial }}
{% set ipArg = "ip=" ~ [node.ip, "" , node.gateway, node.netmask, node.hostname, node.interface, "", node.dns[0], node.dns[1], node.ntp]|join(":") -%}
{% set kernelArgs = [ipArg, node.kernelArgs|join(" "), node.extraKernelArgs|join(" ")]|join(" ") -%}
imgfree
kernel https://pxe.factory.talos.dev/image/{{ .schematicID }}/{{ .talosVersion }}/kernel-{{ .arch }} {{ $kernelArgs }}
initrd https://pxe.factory.talos.dev/image/{{ .schematicID }}/{{ .talosVersion }}/initramfs-{{ .arch }}.xz
kernel https://pxe.factory.talos.dev/image/{{ node.schematicID }}/{{ node.talosVersion }}/kernel-{{ node.arch }} {{ kernelArgs }}
initrd https://pxe.factory.talos.dev/image/{{ node.schematicID }}/{{ node.talosVersion }}/initramfs-{{ node.arch }}.xz
boot
{{- end }}
{{ end }}
{% endif %}
{% endfor %}

18
templates/dnsmasq.conf
View File

@@ -1,4 +1,4 @@
{{ $tftpIp := (ds "config").dhcp.tftpIp -}}
{% set tftpIp = config.dhcp.tftpIp -%}
enable-tftp
tftp-root=/tftproot
@@ -9,9 +9,9 @@ dhcp-vendorclass=UEFI,PXEClient:Arch:00007
dhcp-vendorclass=UEFI64,PXEClient:Arch:00009
# 1st stage: pxe rom boot on ipxe
dhcp-boot=net:BIOS,ipxe.pxe,{{ $tftpIp }},{{ $tftpIp }}
dhcp-boot=net:UEFI,ipxe.efi,{{ $tftpIp }},{{ $tftpIp }}
dhcp-boot=net:UEFI64,ipxe.efi,{{ $tftpIp }},{{ $tftpIp }}
dhcp-boot=net:BIOS,ipxe.pxe,{{ tftpIp }},{{ tftpIp }}
dhcp-boot=net:UEFI,ipxe.efi,{{ tftpIp }},{{ tftpIp }}
dhcp-boot=net:UEFI64,ipxe.efi,{{ tftpIp }},{{ tftpIp }}
# Based on logic in https://gist.github.com/robinsmidsrod/4008017
# iPXE sends a 175 option, checking suboptions
@@ -30,11 +30,11 @@ tag-if=set:ipxe-ok,tag:ipxe-http,tag:ipxe-https
# these create option 43 cruft, which is required in proxy mode
# TFTP IP is required on all dhcp-boot lines (unless dnsmasq itself acts as tftp server?)
pxe-service=tag:!ipxe-ok,X86PC,PXE,undionly.kpxe,{{ $tftpIp }}
pxe-service=tag:!ipxe-ok,IA32_EFI,PXE,snponlyx32.efi,{{ $tftpIp }}
pxe-service=tag:!ipxe-ok,BC_EFI,PXE,snponly.efi,{{ $tftpIp }}
pxe-service=tag:!ipxe-ok,X86-64_EFI,PXE,snponly.efi,{{ $tftpIp }}
pxe-service=tag:!ipxe-ok,X86PC,PXE,undionly.kpxe,{{ tftpIp }}
pxe-service=tag:!ipxe-ok,IA32_EFI,PXE,snponlyx32.efi,{{ tftpIp }}
pxe-service=tag:!ipxe-ok,BC_EFI,PXE,snponly.efi,{{ tftpIp }}
pxe-service=tag:!ipxe-ok,X86-64_EFI,PXE,snponly.efi,{{ tftpIp }}
# later match overrides previous, keep ipxe script last
# server address must be non zero, but can be anything as long as iPXE script is not fetched over TFTP
dhcp-boot=tag:ipxe-ok,boot.ipxe,,{{ $tftpIp }}
dhcp-boot=tag:ipxe-ok,boot.ipxe,,{{ tftpIp }}

119
tools/merge
View File

@@ -1,119 +0,0 @@
#!/usr/bin/env python3
# Adapted from: https://enix.io/en/blog/pxe-talos/
import functools
import json
import pathlib
import requests
import yaml
from jinja2 import Environment, FileSystemLoader, StrictUndefined, Template
NODES = pathlib.Path("nodes")
SCHEMATICS = pathlib.Path("schematics")
PATCHES = Environment(loader=FileSystemLoader("patches"), undefined=StrictUndefined)
TEMPLATES = Environment(loader=FileSystemLoader("templates"), undefined=StrictUndefined)
def node_encoder(node: dict):
class Inner(json.JSONEncoder):
def default(self, o):
if isinstance(o, Template):
try:
rendered = o.render(node)
except Exception as e:
e.add_note(f"While rendering for: {node['hostname']}")
raise e
# Parse the rendered yaml and convert it to a json patch
return json.dumps(yaml.safe_load(rendered))
return super().default(o)
return Inner
@functools.cache
def get_schematic_id(schematic: str):
"""Lookup the schematic id associated with a given schematic"""
r = requests.post("https://factory.talos.dev/schematics", data=schematic)
r.raise_for_status()
data = r.json()
return data["id"]
def schematic_constructor(loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode):
"""Load specified schematic file and get the assocatied schematic id"""
schematic_name = loader.construct_yaml_str(node)
try:
schematic = SCHEMATICS.joinpath(schematic_name).with_suffix(".yaml").read_text()
return get_schematic_id(schematic)
except Exception:
raise yaml.MarkedYAMLError("Failed to load schematic", node.start_mark)
def patch_constructor(loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode):
patch_name = loader.construct_scalar(node)
try:
template = PATCHES.get_template(f"{patch_name}.yaml")
return template
except Exception:
raise yaml.MarkedYAMLError("Failed to load patch", node.start_mark)
def get_loader():
"""Add special constructors to yaml loader"""
loader = yaml.SafeLoader
loader.add_constructor("!schematic", schematic_constructor)
loader.add_constructor("!patch", patch_constructor)
return loader
@functools.cache
def get_defaults(directory: pathlib.Path, root: pathlib.Path):
"""Compute the defaults from the provided directory and parents."""
try:
with open(directory.joinpath("_defaults.yaml")) as fyaml:
yml_data = yaml.load(fyaml, Loader=get_loader())
except OSError:
yml_data = {}
# Stop recursion when reaching root directory
if directory != root:
return get_defaults(directory.parent, root) | yml_data
else:
return yml_data
def walk_files(root: pathlib.Path):
"""Get all files that do not start with and underscore"""
for dirpath, _dirnames, filenames in root.walk():
for fn in filenames:
if not fn.startswith("_"):
yield dirpath.joinpath(fn)
def main():
nodes = []
for fullname in walk_files(NODES):
filename = str(fullname.relative_to(NODES).parent) + "/" + fullname.stem
with open(fullname) as fyaml:
yml_data = yaml.load(fyaml, Loader=get_loader())
yml_data = get_defaults(fullname.parent, NODES) | yml_data
yml_data["hostname"] = fullname.stem
yml_data["filename"] = filename
nodes.append(yml_data)
final_nodes = []
for node in nodes:
# Quick and dirty way to resolve all the templates using a custom encoder
final_nodes.append(json.loads(json.dumps(node, cls=node_encoder(node))))
# Dump everything to json
print(json.dumps(final_nodes, indent=4))
if __name__ == "__main__":
main()

135
tools/render
View File

@@ -1,11 +1,128 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT=$(git rev-parse --show-toplevel)
RENDERED=${ROOT}/rendered
TEMPLATES=${ROOT}/templates
#!/usr/bin/env python3
${ROOT}/tools/merge ./nodes > ${RENDERED}/nodes.json
# Adapted from: https://enix.io/en/blog/pxe-talos/
gomplate --input-dir ${TEMPLATES} --output-dir ${RENDERED} \
-d nodes=file://${RENDERED}/nodes.json \
-d config=${ROOT}/config.yaml \
import functools
import json
import pathlib
import requests
import yaml
from jinja2 import Environment, FileSystemLoader, StrictUndefined, Template
NODES = pathlib.Path("nodes")
SCHEMATICS = pathlib.Path("schematics")
RENDERED = pathlib.Path("rendered")
PATCHES = Environment(loader=FileSystemLoader("patches"), undefined=StrictUndefined)
TEMPLATES = Environment(loader=FileSystemLoader("templates"), undefined=StrictUndefined)
def node_encoder(node: dict):
class Inner(json.JSONEncoder):
def default(self, o):
if isinstance(o, Template):
try:
rendered = o.render(node)
except Exception as e:
e.add_note(f"While rendering for: {node['hostname']}")
raise e
# Parse the rendered yaml and convert it to a json patch
return json.dumps(yaml.safe_load(rendered))
return super().default(o)
return Inner
@functools.cache
def get_schematic_id(schematic: str):
"""Lookup the schematic id associated with a given schematic"""
r = requests.post("https://factory.talos.dev/schematics", data=schematic)
r.raise_for_status()
data = r.json()
return data["id"]
def schematic_constructor(loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode):
"""Load specified schematic file and get the assocatied schematic id"""
schematic_name = loader.construct_yaml_str(node)
try:
schematic = SCHEMATICS.joinpath(schematic_name).with_suffix(".yaml").read_text()
return get_schematic_id(schematic)
except Exception:
raise yaml.MarkedYAMLError("Failed to load schematic", node.start_mark)
def patch_constructor(loader: yaml.SafeLoader, node: yaml.nodes.ScalarNode):
patch_name = loader.construct_scalar(node)
try:
template = PATCHES.get_template(f"{patch_name}.yaml")
return template
except Exception:
raise yaml.MarkedYAMLError("Failed to load patch", node.start_mark)
def get_loader():
"""Add special constructors to yaml loader"""
loader = yaml.SafeLoader
loader.add_constructor("!schematic", schematic_constructor)
loader.add_constructor("!patch", patch_constructor)
return loader
@functools.cache
def get_defaults(directory: pathlib.Path, root: pathlib.Path):
"""Compute the defaults from the provided directory and parents."""
try:
with open(directory.joinpath("_defaults.yaml")) as fyaml:
yml_data = yaml.load(fyaml, Loader=get_loader())
except OSError:
yml_data = {}
# Stop recursion when reaching root directory
if directory != root:
return get_defaults(directory.parent, root) | yml_data
else:
return yml_data
def walk_files(root: pathlib.Path):
"""Get all files that do not start with and underscore"""
for dirpath, _dirnames, filenames in root.walk():
for fn in filenames:
if not fn.startswith("_"):
yield dirpath.joinpath(fn)
def main():
nodes = []
for fullname in walk_files(NODES):
filename = str(fullname.relative_to(NODES).parent) + "/" + fullname.stem
with open(fullname) as fyaml:
yml_data = yaml.load(fyaml, Loader=get_loader())
yml_data = get_defaults(fullname.parent, NODES) | yml_data
yml_data["hostname"] = fullname.stem
yml_data["filename"] = filename
nodes.append(yml_data)
final_nodes = []
for node in nodes:
# Quick and dirty way to resolve all the templates using a custom encoder
final_nodes.append(json.loads(json.dumps(node, cls=node_encoder(node))))
with open("config.yaml") as fyaml:
config = yaml.safe_load(fyaml)
RENDERED.mkdir(exist_ok=True)
for template_name in TEMPLATES.list_templates():
template = TEMPLATES.get_template(template_name)
rendered = template.render(nodes=final_nodes, config=config)
with open(RENDERED.joinpath(template_name), "w") as f:
f.write(rendered)
if __name__ == "__main__":
main()