Add script to manage instance for backup testing

.gitignore

@@ -1,2 +1,3 @@
+**/__pycache__/**
 fact_cache/**
 vault.yml

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)