panxiao81
88a02ababa
Reorganize the brownfield repository, remove retired and generated artifacts, harden ignore rules, and record the GitOps/IaC redesign.
134 lines
6.0 KiB
Python
134 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate imports.tf so Terraform ADOPTS the objects already in NetBox.
|
|
|
|
uv run --with requests --with pyyaml python netbox/terraform/gen-imports.py
|
|
|
|
Run once, when converting an already-populated NetBox to Terraform management. Without it
|
|
the first `terraform apply` tries to CREATE objects that exist and fails on uniqueness.
|
|
|
|
This is a one-shot bootstrap, not part of the normal loop: once `terraform apply` has run,
|
|
state holds the IDs and imports.tf can be deleted. Same intent as
|
|
../../../infrastructure/openbao/terraform/imports.tf.
|
|
|
|
It reads the same topology.yml Terraform does, so the resource addresses and for_each keys
|
|
line up by construction rather than by hand-transcription.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
import yaml
|
|
|
|
HERE = Path(__file__).parent
|
|
for envfile in (HERE / "../seed/.env", HERE / ".env"):
|
|
if envfile.exists():
|
|
for line in envfile.read_text().splitlines():
|
|
if line.strip() and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
URL = os.environ.get("NETBOX_URL", "https://netbox.ad.ddupan.top").rstrip("/")
|
|
TOKEN = os.environ.get("NETBOX_TOKEN", "")
|
|
if not TOKEN:
|
|
sys.exit("NETBOX_TOKEN not set")
|
|
|
|
S = requests.Session()
|
|
S.headers.update({"Authorization": f"Bearer {TOKEN}"})
|
|
|
|
|
|
def one(path: str, **params) -> str | None:
|
|
r = S.get(f"{URL}/api{path}", params={**params, "limit": 1}, timeout=30)
|
|
r.raise_for_status()
|
|
res = r.json()["results"]
|
|
return str(res[0]["id"]) if res else None
|
|
|
|
|
|
def main() -> int:
|
|
t = yaml.safe_load((HERE / "topology.yml").read_text())
|
|
out: list[str] = [
|
|
"# GENERATED by gen-imports.py — one-shot bootstrap, safe to delete after the first",
|
|
"# successful `terraform apply`. Do not hand-edit.",
|
|
"",
|
|
]
|
|
missing: list[str] = []
|
|
|
|
def emit(addr: str, oid: str | None, what: str):
|
|
if oid:
|
|
out.append(f'import {{\n to = {addr}\n id = "{oid}"\n}}\n')
|
|
else:
|
|
missing.append(what)
|
|
|
|
emit("netbox_site.this", one("/dcim/sites/", slug=t["site"]["slug"]), "site")
|
|
emit("netbox_vlan_group.this", one("/ipam/vlan-groups/", slug=t["vlan_group"]["slug"]), "vlan group")
|
|
|
|
for r in t["prefix_roles"]:
|
|
emit(f'netbox_ipam_role.this["{r["slug"]}"]', one("/ipam/roles/", slug=r["slug"]), r["slug"])
|
|
for v in t["vlans"]:
|
|
emit(f'netbox_vlan.this["{v["vid"]}"]', one("/ipam/vlans/", vid=v["vid"]), f"vlan {v['vid']}")
|
|
for p in t["prefixes"]:
|
|
emit(f'netbox_prefix.this["{p["prefix"]}"]', one("/ipam/prefixes/", prefix=p["prefix"]), p["prefix"])
|
|
for r in t["ip_ranges"]:
|
|
emit(f'netbox_ip_range.this["{r["start"]}-{r["end"]}"]',
|
|
one("/ipam/ip-ranges/", start_address=r["start"], end_address=r["end"]), "ip range")
|
|
for w in t["wireless_lans"]:
|
|
emit(f'netbox_wireless_lan.this["{w["ssid"]}"]', one("/wireless/wireless-lans/", ssid=w["ssid"]), w["ssid"])
|
|
for m in t["manufacturers"]:
|
|
emit(f'netbox_manufacturer.this["{m["slug"]}"]', one("/dcim/manufacturers/", slug=m["slug"]), m["slug"])
|
|
for d in t["device_types"]:
|
|
emit(f'netbox_device_type.this["{d["slug"]}"]', one("/dcim/device-types/", slug=d["slug"]), d["slug"])
|
|
for r in t["device_roles"]:
|
|
emit(f'netbox_device_role.this["{r["slug"]}"]', one("/dcim/device-roles/", slug=r["slug"]), r["slug"])
|
|
for c in t["cluster_types"]:
|
|
emit(f'netbox_cluster_type.this["{c["slug"]}"]', one("/virtualization/cluster-types/", slug=c["slug"]), c["slug"])
|
|
for c in t["clusters"]:
|
|
emit(f'netbox_cluster.this["{c["name"]}"]', one("/virtualization/clusters/", name=c["name"]), c["name"])
|
|
|
|
for d in t["devices"]:
|
|
did = one("/dcim/devices/", name=d["name"])
|
|
emit(f'netbox_device.this["{d["name"]}"]', did, d["name"])
|
|
for i in d["interfaces"]:
|
|
key = f'{d["name"]}:{i["name"]}'
|
|
iid = one("/dcim/interfaces/", device_id=did, name=i["name"]) if did else None
|
|
emit(f'netbox_device_interface.this["{key}"]', iid, key)
|
|
if i.get("ip"):
|
|
emit(f'netbox_ip_address.device["{key}"]', one("/ipam/ip-addresses/", address=i["ip"]), i["ip"])
|
|
# netbox_device_primary_ip is deliberately NOT imported: the provider has no
|
|
# importable object at the device ID ("no object exists with the given id").
|
|
# Letting Terraform "create" it simply re-PATCHes primary_ip4 to the value it
|
|
# already holds, which NetBox treats as a no-op.
|
|
if i.get("mac"):
|
|
mac = i["mac"].upper()
|
|
mid = one("/dcim/mac-addresses/", mac_address=mac)
|
|
emit(f'netbox_mac_address.this["{key}"]', mid, mac)
|
|
if iid:
|
|
emit(f'netbox_device_interface_primary_mac_address.this["{key}"]', iid, f"{key} primary mac")
|
|
for it in d.get("inventory_items", []):
|
|
key = f'{d["name"]}:{it["name"]}'
|
|
emit(f'netbox_inventory_item.this["{key}"]',
|
|
one("/dcim/inventory-items/", device_id=did, name=it["name"]) if did else None, key)
|
|
|
|
for v in t["virtual_machines"]:
|
|
vid = one("/virtualization/virtual-machines/", name=v["name"])
|
|
emit(f'netbox_virtual_machine.this["{v["name"]}"]', vid, v["name"])
|
|
for i in v["interfaces"]:
|
|
key = f'{v["name"]}:{i["name"]}'
|
|
iid = one("/virtualization/interfaces/", virtual_machine_id=vid, name=i["name"]) if vid else None
|
|
emit(f'netbox_interface.this["{key}"]', iid, key)
|
|
if i.get("ip"):
|
|
emit(f'netbox_ip_address.vm["{key}"]', one("/ipam/ip-addresses/", address=i["ip"]), i["ip"])
|
|
|
|
(HERE / "imports.tf").write_text("\n".join(out))
|
|
n = sum(1 for line in out if line.startswith("import {"))
|
|
print(f"wrote imports.tf with {n} import blocks")
|
|
if missing:
|
|
print(f"NOT FOUND in NetBox (Terraform will create these): {', '.join(missing)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|