chore: Move things around

imgbb.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3.9
+import argparse
+import concurrent.futures
+import os
+import sys
+import typing
+from pathlib import Path
+from typing import Optional
+
+import httpx
+
+
+IMGBB_TOKEN = os.getenv("IMGBB_TOKEN", "391f0559cc209b11f37eda0084bb3281")
+client = httpx.Client(base_url="https://api.imgbb.com/1/", params={"key": IMGBB_TOKEN})
+
+
+def upload_image(image_path: Path) -> str:
+    filename = f"i{image_path.suffix}"
+    with image_path.open("rb") as f:
+        res = client.post("/upload", files={"image": (filename, f)})
+    res.raise_for_status()
+    data = res.json()["data"]
+    return data["url"]
+
+
+def parse_args(argv: Optional[typing.Sequence[str]] = None) -> argparse.Namespace:
+    arger = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter)
+    arger.add_argument("image_paths", nargs="+", action="store", type=Path, help="Path to image to upload")
+    if not argv or len(argv) == 0:
+        arger.print_help()
+        sys.exit(1)
+
+    return arger.parse_args(argv)
+
+
+def main():
+    args = parse_args(sys.argv[1:])
+    image_paths = args.image_paths
+    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
+        results = list(pool.map(upload_image, image_paths))
+    for url in results:
+        print(url)
+
+
+if __name__ == "__main__":
+    main()

legacy/hetzner_kube.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3.9
+import base64
+import json
+import logging
+import pprint
+import subprocess
+import time
+import webbrowser
+from pathlib import Path
+
+import httpx
+import typer
+
+hetzner_token = 'zdlkmAF9EswvOIEGvWgutULYTmUpa3c44YcKZOB0jymQfgvra7kRVJxuLiFTouM9'
+client = httpx.Client(base_url='https://api.hetzner.cloud/v1', headers={'Authorization': f'Bearer {hetzner_token}'})
+
+
+datacenters = [
+    {'id': 2, 'name': 'nbg1-dc3', 'description': 'Nuremberg 1 DC 3'},
+    {'id': 3, 'name': 'hel1-dc2', 'description': 'Helsinki 1 DC 2'},
+    {'id': 4, 'name': 'fsn1-dc14', 'description': 'Falkenstein 1 DC14'},
+    {'id': 5, 'name': 'ash-dc1', 'description': 'Ashburn DC1'},
+]
+
+server_types = {
+    'cx11': {'cores': 1, 'memory': 2.0, 'disk': 20},
+    'cx21': {'cores': 2, 'memory': 4.0, 'disk': 40},
+    'cx31': {'cores': 2, 'memory': 8.0, 'disk': 80},
+    'cx41': {'cores': 4, 'memory': 16.0, 'disk': 160},
+    'cx51': {'cores': 8, 'memory': 32.0, 'disk': 240},
+    'ccx11': {'cores': 2, 'memory': 8.0, 'disk': 80},
+    'ccx21': {'cores': 4, 'memory': 16.0, 'disk': 160},
+    'ccx31': {'cores': 8, 'memory': 32.0, 'disk': 240},
+    'ccx41': {'cores': 16, 'memory': 64.0, 'disk': 360},
+    'ccx51': {'cores': 32, 'memory': 128.0, 'disk': 600},
+    'cpx11': {'cores': 2, 'memory': 2.0, 'disk': 40},
+    'cpx21': {'cores': 3, 'memory': 4.0, 'disk': 80},
+    'cpx31': {'cores': 4, 'memory': 8.0, 'disk': 160},
+    'cpx41': {'cores': 8, 'memory': 16.0, 'disk': 240},
+    'cpx51': {'cores': 16, 'memory': 32.0, 'disk': 360},
+    'ccx12': {'cores': 2, 'memory': 8.0, 'disk': 80},
+    'ccx22': {'cores': 4, 'memory': 16.0, 'disk': 160},
+    'ccx32': {'cores': 8, 'memory': 32.0, 'disk': 240},
+    'ccx42': {'cores': 16, 'memory': 64.0, 'disk': 360},
+    'ccx52': {'cores': 32, 'memory': 128.0, 'disk': 600},
+    'ccx62': {'cores': 48, 'memory': 192.0, 'disk': 960},
+}
+
+
+def create_server(name: str, datacenter: str):
+    payload = {
+        "datacenter": datacenter,
+        "image": "ubuntu-20.04",
+        "name": name,
+        "server_type": "cpx21",
+        "ssh_keys": ["entropy"],
+    }
+
+    res = client.post('/servers', json=payload)
+    res.raise_for_status()
+
+    data = res.json()
+    server = data['server']
+    pprint.pprint(server)
+
+    return {
+        'id': server['id'],
+        'name': server['name'],
+        'ip': server['public_net']['ipv4']['ip'],
+    }
+
+
+def find_server(name: str) -> dict:
+    res = client.get('/servers')
+    res.raise_for_status()
+
+    data = res.json()
+
+    server = [it for it in data['servers'] if it['name'] == name][0]
+    return {
+        'id': server['id'],
+        'name': server['name'],
+        'ip': server['public_net']['ipv4']['ip'],
+    }
+
+
+def delete_server(name: str) -> None:
+    server = find_server(name)
+    logging.info(f'deleting server={name} with id={server["id"]}')
+
+    res = client.delete(f'/servers/{server["id"]}')
+    res.raise_for_status()
+
+    logging.info(f'deleted server={name}')
+
+
+def copy_to_clipboard(text: str) -> None:
+    subprocess.Popen(['pbcopy'], stdin=subprocess.PIPE).communicate(input=text.encode())
+
+
+class Server:
+    ssh_options = ['-o', 'StrictHostKeyChecking=no']
+
+    def __init__(self, ip_address: str, name: str = None):
+        self.ip = ip_address
+        self.ssh_host = f'root@{self.ip}'
+        self.name = name
+
+    @classmethod
+    def from_hetzner(cls, server: dict):
+        return cls(server['ip'], name=server['name'])
+
+    def run_command(self, cmd: str, return_output: bool = False) -> str:
+
+        logging.debug(f'running command={cmd} on {self.ssh_host}')
+        extra_args = {}
+        if return_output:
+            extra_args['stdout'] = subprocess.PIPE
+        p = subprocess.run(['ssh', *self.ssh_options, self.ssh_host, '--', cmd], text=True, **extra_args)
+        p.check_returncode()
+        return p.stdout
+
+    def install_package(self, packages: str) -> str:
+        return self.run_command(f'apt-get install -y {packages}')
+
+    def install_shell(self):
+        self.install_package('fish')
+        self.run_command('chsh -s /usr/bin/fish')
+
+    def install_microk8s(self):
+        self.install_package('snapd')
+        self.run_command('snap install microk8s --classic')
+        self.run_command('ufw allow in on cni0 && ufw allow out on cni0 && ufw default allow routed')
+        self.run_command('microk8s enable dashboard dns rbac ingress storage prometheus registry')
+
+    def get_microk8s_token(self):
+        raw_json = self.run_command('microk8s kubectl -n kube-system get secret -o json', return_output=True)
+        secrets = json.loads(raw_json)['items']
+
+        default_token_secret = [it for it in secrets if it['metadata']['name'].startswith('default-token')][0]
+        token_base64 = default_token_secret['data']['token']
+
+        token = base64.decodebytes(token_base64.encode()).decode()
+        return token
+
+    def show_microk8s_access_info(self):
+        logging.info('getting default service account token')
+        token = self.get_microk8s_token()
+        print(token)
+        logging.info('copying token to clipboard')
+        copy_to_clipboard(token)
+
+        logging.info('getting dashboard ip')
+        proxy = self.get_microk8s_dashboard_proxy()
+        print(proxy['ssh'])
+        copy_to_clipboard(proxy['ssh'])
+
+        logging.info('opening url in the browser')
+        print(proxy['url'])
+        webbrowser.open_new_tab(proxy['url'])
+
+    def get_microk8s_dashboard_proxy(self):
+        cluster_ip = self.run_command(
+            'microk8s kubectl get -n kube-system -o json service/kubernetes-dashboard -o jsonpath="{.spec.clusterIP}"',
+            return_output=True,
+        ).strip()
+        port, local_port = 443, 18433
+
+        return {
+            'ssh': f'ssh -vNL {local_port}:{cluster_ip}:{port} {self.ssh_host}',
+            'url': f'https://localhost:{local_port}',
+        }
+
+    def write_kubeconfig(self, filename: str):
+        logging.info(f'writing kubeconfig to {filename}')
+        kubeconfig = self.run_command('microk8s config', return_output=True)
+        file_path = Path(filename).expanduser()
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        file_path.write_text(kubeconfig)
+
+    def install_micro_editor(self):
+        logging.info('installing micro editor')
+        self.run_command('curl https://getmic.ro | bash && mv ./micro /usr/bin/')
+
+    def setup(self):
+        self.run_command('apt-get update')
+
+        self.install_shell()
+        self.install_micro_editor()
+        self.install_microk8s()
+
+        logging.info('waiting until pods are online')
+        time.sleep(10)
+
+        self.show_microk8s_access_info()
+        self.write_kubeconfig(f'~/Desktop/kubeconfig-{self.name}.yml')
+
+
+cli = typer.Typer(invoke_without_command=False)
+
+
+@cli.command()
+def create(
+    name: str,
+    datacenter: str = typer.Option('nbg1-dc3'),
+):
+    create_server(name=name, datacenter=datacenter)
+
+
+@cli.command()
+def destroy(
+    name: str,
+):
+    delete_server(name=name)
+
+
+@cli.command()
+def setup(
+    name: str,
+):
+    server = find_server(name)
+    server = Server.from_hetzner(server)
+    server.setup()
+
+
+@cli.command()
+def show_k8s(
+    name: str,
+):
+    server = Server.from_hetzner(find_server(name))
+    server.show_microk8s_access_info()
+    server.write_kubeconfig(f'~/Desktop/kubeconfig-{server.name}.yml')
+
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.INFO)
+    cli()