From 5be55541cbc4d8a3254814497fd453998708e511 Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Sun, 26 Jan 2025 23:22:36 +0300 Subject: [PATCH] chore: Move things around --- imgbb.py | 46 ++++++++ legacy/hetzner_kube.py | 237 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 283 insertions(+) create mode 100755 imgbb.py create mode 100755 legacy/hetzner_kube.py diff --git a/imgbb.py b/imgbb.py new file mode 100755 index 0000000..9f76334 --- /dev/null +++ b/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() diff --git a/legacy/hetzner_kube.py b/legacy/hetzner_kube.py new file mode 100755 index 0000000..1b0f77b --- /dev/null +++ b/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()