diff --git a/.gitignore b/.gitignore index 59395c1..ae8d709 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ +**/__pycache__/** fact_cache/** vault.yml diff --git a/scripts/scaleway/baldur.py b/scripts/scaleway/baldur.py new file mode 100644 index 0000000..77426ec --- /dev/null +++ b/scripts/scaleway/baldur.py @@ -0,0 +1,146 @@ +import argparse +import keyring +import requests + + +class Scaleway: + API_ENDPOINT_BASE = "https://api.scaleway.com/instance/v1/zones" + ZONES = [ + "fr-par-1", "fr-par-2", "fr-par-3", + "nl-ams-1", "nl-ams-2", + "pl-waw-1", "pl-waw-2", + ] + + def __init__(self, project_id, secret_key): + self.__zone = None + self.__project_id = project_id + self.__headers = {"X-Auth-Token": secret_key} + + @property + def zone(self): + return self.__zone + + @zone.setter + def zone(self, zone): + if zone not in Scaleway.ZONES: + raise KeyError(f"{zone} is not a valid zone - must be one of {Scaleway.ZONES}") + self.__zone = zone + + @property + def project_id(self): + return self.__project_id + + def __url(self, item, id): + if self.__zone is None: + raise RuntimeError("zone must be set before making any API requests") + + url = f"{Scaleway.API_ENDPOINT_BASE}/{self.__zone}" + + if id == "products": + return f"{url}/products/{item}" + + url = f"{url}/{item}" + if id is not None: + url = f"{url}/{id}" + + return url + + @staticmethod + def __check_status(type, url, rsp): + if (rsp.status_code // 100) != 2: + raise RuntimeError( + f"{type} {url} returned with status code {rsp.status_code}: {rsp.json()}") + + def get(self, item, id=None): + url = self.__url(item, id) + r = requests.get(url, headers=self.__headers) + self.__check_status("GET", url, r) + return r.json()[item] + + def get_by_name(self, item, name): + items = self.get(item) + return next((it for it in items if it["name"] == name), None) + + def __post(self, url, data): + r = requests.post(url, headers=self.__headers, json=data) + self.__check_status("POST", url, r) + return r.json() + + def post(self, item, data): + return self.__post(self.__url(item, None), data) + + def post_action(self, item, id, action, data): + return self.__post(f"{self.__url(item, id)}/{action}", data) + + def delete(self, item, id): + url = self.__url(item, id) + r = requests.delete(url, headers=self.__headers) + self.__check_status("DELETE", url, r) + + +def create_baldur(scaleway, args): + volume_size = args.volume_size + + security_group = scaleway.get_by_name("security_groups", "baldur-security-group") + image = scaleway.get_by_name("images", "Debian Bullseye") + server_type = "PLAY2-PICO" + if server_type not in scaleway.get("servers", id="products"): + raise RuntimeError(f"{server_type} is not available in {scaleway.zone}") + + response = scaleway.post("ips", data={"project": scaleway.project_id}) + public_ip = response["ip"] + + baldur = { + "name": "baldur", + "dynamic_ip_required": False, + "commercial_type": server_type, + "image": image["id"], + "volumes": {"0": {"size": int(volume_size * 1_000_000_000)}}, + "enable_ipv6": False, + "public_ip": public_ip["id"], + "project": scaleway.project_id, + "security_group": security_group["id"], + } + + response = scaleway.post("servers", data=baldur) + server = response["server"] + + scaleway.post_action("servers", server["id"], "action", data={"action": "poweron"}) + + print("Baldur instance created:") + print(f" block volume size: {server['volumes']['0']['size']//1_000_000_000} GiB") + print(f" public ip address: {server['public_ip']['address']}") + + +def delete_baldur(scaleway, _): + server = scaleway.get_by_name("servers", "baldur") + if server is None: + raise RuntimeError(f"Baldur instance was not found in {scaleway.zone}") + ip = server["public_ip"] + + scaleway.post_action("servers", server["id"], "action", data={"action": "terminate"}) + scaleway.delete("ips", ip["id"]) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Create or delete the Baldur instance") + + subparsers = parser.add_subparsers() + + create_parser = subparsers.add_parser("create") + create_parser.add_argument("--volume-size", type=int, required=True, + help="Block volume size (in GiB) to create") + create_parser.set_defaults(func=create_baldur) + + delete_parser = subparsers.add_parser("delete") + delete_parser.set_defaults(func=delete_baldur) + + args = parser.parse_args() + + scw_project_id = keyring.get_password("scaleway", "project_id") + scw_secret_key = keyring.get_password("scaleway", "secret_key") + + scaleway = Scaleway(scw_project_id, scw_secret_key) + scaleway.zone = "fr-par-2" + + args.func(scaleway, args)