lab/labsim/sim-ha-config.py

#!/usr/bin/env python3
"""Generate the HA config for the labsim VyOS pair.

Exists to answer one question that cannot be answered on a single router, and
that would otherwise only be discovered at cutover: with kea HA active-passive,
does exactly ONE box answer a DHCP request?

Mirrors the production shape so the answer transfers:

    router1  172.31.<v>.252   priority 200   DHCP HA primary
    router2  172.31.<v>.253   priority 100   DHCP HA secondary
    VIP      172.31.<v>.1     (what clients use as their gateway)

Note the sim's LoT VLAN is a /23 like production, so the VIP prefix differs
there -- getting that wrong produces a config that commits and then behaves
subtly wrongly, which is worse than a failure.

  ./sim-ha-config.py --role primary   > r1.conf
  ./sim-ha-config.py --role secondary > r2.conf
"""
from __future__ import annotations

import argparse
import importlib.util
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
MIG = os.path.join(HERE, "..", "migration")

# Reuse the DHCP/DNS generator rather than hand-writing subnets: the whole
# point is that what is proven here and what production gets share a code path.
_spec = importlib.util.spec_from_file_location(
    "unifi_to_vyos", os.path.join(MIG, "unifi-to-vyos.py"))
unifi_to_vyos = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(unifi_to_vyos)

# vlan -> (prefix, cidr). LoT is a /23 in the sim, matching production.
VLANS = {
    1:   ("172.31.1",   24),
    2:   ("172.31.2",   24),
    3:   ("172.31.3",   24),
    9:   ("172.31.9",   24),
    10:  ("172.31.10",  23),
    200: ("172.31.200", 24),
}
DHCP_HA_NAME = "labsim-dhcp-pair"     # must not equal either host-name


def group(vlan: int) -> str:
    return "native" if vlan == 1 else f"vlan{vlan}"


def build(role: str) -> list[str]:
    primary = role == "primary"
    self_o, peer_o = (252, 253) if primary else (253, 252)
    prio = 200 if primary else 100
    out = [f"# labsim VyOS HA -- {role}", ""]

    for vlan, (pfx, cidr) in VLANS.items():
        g = group(vlan)
        iface = "bond0" if vlan == 1 else f"bond0 vif {vlan}"
        out += [
            f"# VLAN {vlan}",
            # The node's own address replaces the .1 it used to hold directly;
            # .1 becomes the floating VIP, exactly as production will be.
            f"delete interfaces bonding {iface} address",
            f"set interfaces bonding {iface} address '{pfx}.{self_o}/{cidr}'",
            f"set high-availability vrrp group {g} interface bond0{'' if vlan == 1 else f'.{vlan}'}",
            f"set high-availability vrrp group {g} vrid {vlan}",
            f"set high-availability vrrp group {g} address {pfx}.1/{cidr}",
            f"set high-availability vrrp group {g} priority {prio}",
            f"set high-availability vrrp group {g} hello-source-address {pfx}.{self_o}",
            f"set high-availability vrrp group {g} peer-address {pfx}.{peer_o}",
            f"set high-availability vrrp group {g} no-preempt",
            f"set high-availability vrrp sync-group MAIN member {g}",
            "",
        ]

    out += [
        "# --- DHCP high-availability ---",
        "# The thing under test: active-passive should mean exactly one OFFER.",
        "set service dhcp-server high-availability mode active-passive",
        f"set service dhcp-server high-availability status {role}",
        f"set service dhcp-server high-availability name {DHCP_HA_NAME}",
        f"set service dhcp-server high-availability source-address 172.31.10.{self_o}",
        f"set service dhcp-server high-availability remote 172.31.10.{peer_o}",
        "",
    ]

    inv = json.load(open(os.path.join(MIG, "export", "inventory.json")))
    dhcp, stats = unifi_to_vyos.build(inv, "sim")
    out += [l for l in dhcp if l.strip() and not l.startswith("#")]
    print(f"{role}: {stats['subnets']} subnets, {stats['mappings']} mappings",
          file=sys.stderr)
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--role", choices=("primary", "secondary"), required=True)
    args = ap.parse_args()
    sys.stdout.write("\n".join(build(args.role)) + "\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())