DPZC/deploy_hetzner.py

from hcloud import Client
from hcloud.networks.domain import NetworkSubnet
from hcloud.locations.domain import Location
from hcloud.server_types.domain import ServerType
from hcloud.images.domain import Image
from hcloud.volumes.domain import Volume
from hcloud.load_balancer_types.domain import LoadBalancerType
from hcloud.load_balancers.domain import LoadBalancerAlgorithm, LoadBalancerServiceHttp, LoadBalancerHealthCheck, LoadBalancerService, LoadBalancerHealtCheckHttp, LoadBalancerTarget

class HetzerDeployment:

    KEY_NAME = "patrycjalazna@patrycjalazna.local"
    client = None
    ssh_key = None


    def __init__(self, name, server_type = "cx11", image = "ubuntu-20.04", location = "hel1"):
        self.name = name
        self.server_type = server_type
        self.image = image
        self.location = location

        self.client = self.create_client()
        self.ssh_key = self.get_ssh_key()



    def get_ssh_key(self):
        with open("ssh_key.txt", "r") as f:
            ssh_pub_key = f.read()

        if self.client.ssh_keys.get_by_name(self.KEY_NAME):
            ssh_key = self.client.ssh_keys.get_by_name(self.KEY_NAME)

        else:
            ssh_key = self.client.ssh_keys.create(name=self.KEY_NAME, public_key=ssh_pub_key)

        return ssh_key


    def get_auth_token(self):
        with open("API_key.txt", "r") as f:
            API_key = f.read()

        return API_key

    def create_client(self):
        API_key = self.get_auth_token()

        client = Client(token=API_key)

        return client


    def create_network(self):

        vnet =  self.client.networks.create(
            name= self.name + "-vnet", 
            ip_range="10.10.10.0/24", 
            subnets=[
            NetworkSubnet(ip_range="10.10.10.0/24", network_zone="eu-central", type="cloud")
                ]
            )
    
        print(f"Utworzono sieć wirtualną {vnet.data_model.name} ({vnet.data_model.ip_range})")

        return vnet




    def create_server(self, vnet, server_name, filename, num, volume=None):

        with open(filename, "r") as f:
            cloud_init = f.read()
        
        name = self.name + server_name + str(num)

        if volume is not None:
            server = self.client.servers.create(
                name=name, 
                server_type=ServerType(self.server_type), 
                image=Image(name=self.image), 
                ssh_keys=[self.ssh_key], 
                networks=[vnet], 
                location=Location(self.location), 
                user_data=cloud_init,
                volumes=[volume.volume]
            )

        else:
             server = self.client.servers.create(
                name=name, 
                server_type=ServerType(self.server_type), 
                image=Image(name=self.image), 
                ssh_keys=[self.ssh_key], 
                networks=[vnet], 
                location=Location(self.location), 
                user_data=cloud_init
            )

        server.action.wait_until_finished()

        print(f"Tworzenie serwera {name}: {server.action.complete}")

        server_ip = self.client.servers.get_by_name(name)
        print(f"Serwer: {server_ip.data_model.name}\n\tpubliczne IP: {server_ip.data_model.public_net.ipv4.ip}")

        return server.server


    def create_load_balancer(self, servers, vnet, lb_type):
        name = self.name + "-load-balancer"

        targets = []
        for s in servers:
            targets.append(LoadBalancerTarget(
                type="server",
                server=s,
                use_private_ip=True
        ))

        load_balancer = self.client.load_balancers.create(
            name=name,
            load_balancer_type=LoadBalancerType(name=lb_type),
            location=Location(self.location),
            services=[
                     LoadBalancerService(
                         protocol="http",
                         listen_port=8080,
                         destination_port=8080,
                         proxyprotocol=False,
                         health_check=LoadBalancerHealthCheck(
                             protocol="http",
                             port="8080",
                             interval=15,
                             timeout=10,
                             retries=3,
                             http=LoadBalancerHealtCheckHttp(
                                 path="/factors/10",
                                 status_codes=["2??", "3??"],
                                 tls=False
                             )
                         )
                     )
                 ],
            targets=targets,
            public_interface=True,
            network=vnet
        )


        load_balancer.action.wait_until_finished()
        print(f"Tworzenie load balancera {name}: {load_balancer.action.complete}")

        load_balancer = self.client.load_balancers.get_by_name(name)
        print(f"Load balancer: {load_balancer.data_model.name}\n\tpubliczne IP: {load_balancer.data_model.public_net.ipv4.ip}")

        return load_balancer


    def deploy(self):
        vnet = self.create_network()
        servers = []
        for i in range(2):
            servers.append(self.create_server(vnet, "prime", "cloud_init.yaml", i))

        load_balancer = self.create_load_balancer(servers, vnet, "lb11")
        

def main():
    h = HetzerDeployment('patlaz---')
    h.deploy()

if __name__ == '__main__':
     main()