ansible-edda/scripts/scaleway/baldur.py

147 lines
4.6 KiB
Python

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)