Cian-H/I-Form_Server_Node_Deployer
1
0
Fork 0
mirror of https://github.com/Cian-H/I-Form_Server_Node_Deployer.git synced 2025-12-23 22:52:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

All functions are now typed and documented

Browse Source
This commit is contained in:
Cian Hughes
2023-11-01 15:11:02 +00:00
parent cba4743035
commit 35fc76054b
11 changed files with 267 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/node_deployer/__main__.py
View File

@@ -1,19 +1,23 @@
#!/usr/bin/env python
def main() -> None:
"""Entry point for the CLI
"""
from . import config
config.update_config("cli")
from .node_deployer import app
app()
def debug():
def debug() -> None:
"""Entry point for the debug CLI
"""
from . import config
config.update_config("debug")
from .node_deployer import app
# Below, we set the default value of the debug flag
# for the base function of each command to True
def unwrap(f):
def unwrap(f): # Not a closure, just here to avoid polluting the namespace
if hasattr(f, "__wrapped__"):
return unwrap(f.__wrapped__)
else:

47
src/node_deployer/autoignition.py
View File

@@ -5,6 +5,7 @@ import tarfile
import time
from typing import Annotated
import docker
import git
from selenium import webdriver
from selenium.webdriver.common.by import By
@@ -18,7 +19,12 @@ from .debug import debug_guard
from .utils import ensure_build_dir
def create_driver():
def create_driver() -> webdriver.Remote:
"""Creates a selenium webdriver instance
Returns:
webdriver.Remote: The created webdriver instance
"""
driver = webdriver.Remote(
"http://127.0.0.1:4444",
options=webdriver.FirefoxOptions(),
@@ -27,7 +33,20 @@ def create_driver():
return driver
def convert_json_via_fuelignition(container, driver, fuelignition_json, img_path):
def convert_json_via_fuelignition(
container: docker.models.containers.Container,
driver: webdriver.Remote,
fuelignition_json: Path,
img_path: Path,
) -> None:
"""Converts a fuel-ignition json file to an ignition disk image file
Args:
container (docker.models.containers.Container): The selenium container
driver (webdriver.Remote): The selenium webdriver instance
fuelignition_json (Path): The path to the fuel-ignition json file
img_path (Path): The path to the output ignition disk image file
"""
driver.get(config.FUELIGNITION_URL)
# Navigate to "Load Settings from" and upload the json
load_from = driver.find_element(By.NAME, "load_from")
@@ -66,7 +85,12 @@ def convert_json_via_fuelignition(container, driver, fuelignition_json, img_path
f.write(tar.extractfile(tar.getmembers()[0].name).read())
def build_fuelignition():
def build_fuelignition() -> docker.models.images.Image:
"""Builds the fuel-ignition docker image
Returns:
docker.models.images.Image: The built docker image
"""
# Make sure the local fuel-ignition repo is up to date
if (not config.FUELIGNITION_BUILD_DIR.exists()) or (
len(tuple(config.FUELIGNITION_BUILD_DIR.iterdir())) == 0
@@ -149,7 +173,22 @@ def json_to_img(
)
] = False,
) -> None:
"""Takes a fuel-ignition json file and produces an ignition disk image file"""
"""Converts a fuel-ignition json file to an ignition disk image file
Args:
json_path (Annotated[ Path, typer.Option, optional):
The path to the fuel-ignition json file.
Defaults to Path("fuelignition.json").
img_path (Annotated[ Path, typer.Option, optional):
The path to the output ignition disk image file.
Defaults to Path("ignition.img").
debug (Annotated[ bool, typer.Option, optional):
Enable debug mode.
Defaults to False.
Raises:
e: Any exception raised during execution
"""
selenium_container = None
fuelignition_container = None
fuelignition_image = None

10
src/node_deployer/cli.py
View File

@@ -9,10 +9,20 @@ from .utils import Singleton
class SingletonProgress(Progress, metaclass=Singleton):
"""A singleton progress bar"""
pass
def cli_spinner(*spinner_args, **spinner_kwargs) -> Callable:
"""A decorator that adds a spinner to the CLI while the decorated function is running
Args:
*spinner_args: The arguments to pass to the rich spinner object
**spinner_kwargs: The keyword arguments to pass to the rich spinner object
Returns:
Callable: The decorated function
"""
def decorator(f: Callable) -> Callable:
# Indent the spinner to match its nesting level
indent = len(inspect.stack()) - 1

41
src/node_deployer/config.py
View File

@@ -17,7 +17,17 @@ PROJECT_ROOT: Path = Path(__file__).parent.parent.parent.absolute()
type ConfigLabel = str | list[str]
def get_config(config_label: ConfigLabel = ["default"]) -> dict:
def get_config(config_label: ConfigLabel = "default") -> dict:
"""Gets the specified configuration from config.toml
Args:
config_label (ConfigLabel, optional):
The label of the configuration to get.
Defaults to "default".
Returns:
dict: The specified configuration
"""
if isinstance(config_label, str):
config_label = [config_label]
with open(PROJECT_ROOT / "config.toml", "rb") as f:
@@ -29,6 +39,12 @@ def get_config(config_label: ConfigLabel = ["default"]) -> dict:
def finalise_config(config: dict) -> None:
"""Finalises the configuration by converting paths to Path objects and
appropriately setting secondary parameters such as relative paths
Args:
config (dict): The configuration to finalise
"""
# First, convert base paths to Path objects
for k, v in config.items():
match k:
@@ -57,17 +73,34 @@ def finalise_config(config: dict) -> None:
def apply_config(config: dict) -> None:
"""Applies the specified configuration to this module's globals
Args:
config (dict): The configuration to apply
"""
finalise_config(config)
globals().update(config)
def update_config(config_label: ConfigLabel = "default") -> None:
"""Updates the configuration to the specified configuration
Args:
config_label (ConfigLabel, optional):
The label of the configuration to update to.
Defaults to "default".
"""
apply_config(get_config(config_label))
def init(config) -> None:
globals().update(get_config(config))
def init(config_label: ConfigLabel) -> None:
"""Initialises the configuration module
Args:
config_label (ConfigLabel): The configuration to initialise with
"""
globals().update(get_config(config_label))
update_config()
init(config="default")
init(config_label="default")

56
src/node_deployer/create_disk.py
View File

@@ -4,6 +4,7 @@ from typing import Annotated
from docker.types import Mount
import typer
from typing import Tuple
from . import config
from .cli import cli_spinner
@@ -16,6 +17,14 @@ type IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
def filter_validation_response(response: str) -> str:
"""Filters out erroneous warnings from the validation response
Args:
response (str): The response to filter
Returns:
str: The filtered response
"""
return "\n".join(
filter(
# Filter out the warning about unused key human_readable, this always exists in
@@ -27,6 +36,11 @@ def filter_validation_response(response: str) -> str:
def validation_result() -> str:
"""Returns the response resulting from a validation of the ignition image
Returns:
str: The response from the validation
"""
dockerfile = config.DOCKERFILE_DIR / "validate.dockerfile"
image, _ = config.CLIENT.images.build(
path=".",
@@ -50,7 +64,13 @@ def validation_result() -> str:
@cli_spinner(description="Validating ignition image", total=None)
def validate() -> (bool, str):
def validate() -> Tuple[bool, str]:
"""Validates the ignition image
Returns:
Tuple[bool, str]: A tuple containing a boolean indicating whether
the validation was successful and the response from the validation
"""
response = validation_result().decode()
response = filter_validation_response(response)
return (not bool(response), response)
@@ -58,6 +78,11 @@ def validate() -> (bool, str):
@cli_spinner(description="Writing ignition image to disk", total=None)
def write_disk(disk: str) -> None:
"""Writes the ignition image to the specified disk
Args:
disk (str): The disk to write to
"""
config.CLIENT.containers.run(
"alpine",
mounts=[config.CWD_MOUNT, Mount("/ignition_disk", disk, type="bind")],
@@ -141,7 +166,34 @@ def create_ignition_disk(
)
] = False,
) -> None:
"""Writes an ignition image to the specified disk for easy deployment of new nodes to the swarm""" # noqa
"""Creates an ignition image and writes it to the specified disk
Args:
disk (Annotated[ str, typer.Option, optional):
The disk to write to.
Defaults to None.
hostname (Annotated[ str, typer.Option, optional):
The hostname for the new node.
Defaults to "node".
password (Annotated[ str, typer.Option, optional):
The password for the root user on the new node.
Defaults to None.
switch_ip (Annotated[ IPAddress, typer.Option, optional):
The IP address of the switch to connect to.
Defaults to None.
switch_port (Annotated[ int, typer.Option, optional):
The port on the switch to connect to.
Defaults to 4789.
swarm_token (Annotated[ str, typer.Option, optional):
The swarm token for connecting to the swarm.
Defaults to None.
debug (Annotated[ bool, typer.Option, optional):
Enable debug mode.
Defaults to False.
Raises:
typer.Exit: Exit CLI if the ignition image is invalid
"""
create_img(
hostname = hostname,
password = password,

42
src/node_deployer/create_img.py
View File

@@ -16,6 +16,11 @@ type IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address
def load_template() -> dict:
"""Loads the default template for the ignition configuration
Returns:
dict: The default ignition configuration
"""
with open(config.SRC_DIR / "templates/fuelignition.json", "r") as f:
out = json.load(f)
return out
@@ -27,6 +32,17 @@ def apply_ignition_settings(
password: str,
swarm_config: str,
) -> dict:
"""Applies the specified ignition settings to the template
Args:
template (dict): The template to apply the settings to
hostname (str): The hostname to set
password (str): The password to set for the root user
swarm_config (str): The swarm configuration to set
Returns:
dict: The template with the settings applied
"""
ignition_config = template.copy()
ignition_config["hostname"] = hostname
ignition_config["login"]["users"][0]["passwd"] = password
@@ -144,7 +160,31 @@ def create_img(
),
] = False,
) -> None:
"""Creates an ignition image for deploying a new node to the swarm"""
"""Creates an ignition image for a node that will automatically join a swarm
Args:
hostname (Annotated[ str, typer.Option, optional):
The hostname to set for the node.
Defaults to "node".
password (Annotated[ str, typer.Option, optional):
The password to set for the root user on the node.
Defaults to None.
switch_ip (Annotated[ IPAddress, typer.Option, optional):
The IP address of the switch to connect to.
Defaults to None.
switch_port (Annotated[ int, typer.Option, optional):
The port on the switch to connect to.
Defaults to 4789.
swarm_token (Annotated[ str, typer.Option, optional):
The swarm token for connecting to the swarm.
Defaults to None.
img_path (Annotated[ Path, typer.Option, optional):
The path to which the ignition image should be written.
Defaults to Path("ignition.img").
debug (Annotated[ bool, typer.Option, optional):
Enable debug mode.
Defaults to False.
"""
# get swarm configuration as JSON
swarm_config = json.dumps(
{

17
src/node_deployer/debug.py
View File

@@ -7,13 +7,18 @@ import typer
from . import config
# def merge(f1: Callable) -> Callable:
# https://docs.python.org/3/library/functools.html#functools.update_wrapper
# wraps, but it combines the signatures of the two functions
# This will allow us to add/remove the `debug` arg depending on config context
def debug_guard(f: Callable) -> Callable:
"""A decorator that contextually enables debug mode for the decorated function
Args:
f (Callable): The function to decorate
Raises:
typer.Exit: Exit CLI if the dev group is not installed
Returns:
Callable: The decorated function
"""
if not config.DEBUG:
return f
try:

15
src/node_deployer/utils.py
View File

@@ -6,6 +6,14 @@ from . import config
def ensure_build_dir(f: Callable) -> Callable:
"""Ensures that the build directory exists before running the decorated function
Args:
f (Callable): The function to decorate
Returns:
Callable: The decorated function
"""
@wraps(f)
def wrapper(*args, **kwargs):
Path(config.BUILD_DIR).mkdir(exist_ok=True, parents=True)
@@ -15,8 +23,15 @@ def ensure_build_dir(f: Callable) -> Callable:
class Singleton(type):
"""A singleton metaclass"""
_instance = None
def __call__(cls, *args, **kwargs):
"""Creates a new instance of the class if one does not already exist
Returns:
cls._instance: The instance of the class
"""
if cls._instance is None:
cls._instance = super().__call__(*args, **kwargs)
return cls._instance