Hetzner Cloud alias IP failover using keepalived

hcloud-failover.py

#!/usr/bin/env python3
# hcloud-failover.py
# Inspired by https://github.com/lehuizi/hcloud-failover-keepalived
# This script is to be called from keepalived on the notify action
# and causes the alias IPs to be updated for the configured network,
# possibly promoting a current backup to master.
# The arguments are
#  = Type ("GROUP" or "INSTANCE")
#  = Name of group or instance
#  = Target state of transition ("MASTER", "BACKUP", "FAULT")
# Example config /etc/hcloud-failover/kube-masters.json
# {
#   "floating_interface": "eth0",
#   "floating_ips": ["my-floating-ip"],
#   "alias_ips": ["10.240.10.10"],
#   "alias_interface": "enp7s0",
#   "api_token": "<hcloud-api-token>",
#   "network": "kubernetes",
#   "servers": ["master01", "master02"]
# }
# Example: ./hcloud-failover.py INSTANCE kube-masters MASTER

import json
import os
import re
import requests
import socket
import sys
import tempfile
import syslog

def log(message):
    syslog.syslog(message)
    print(message)

def get_servers(headers):
    r = requests.get("https://api.hetzner.cloud/v1/servers", headers=headers)
    return r.json()["servers"]

def get_floating_ips(headers):
    r = requests.get("https://api.hetzner.cloud/v1/floating_ips", headers=headers)
    return r.json()["floating_ips"]

def get_network(headers, name):
    r = requests.get("https://api.hetzner.cloud/v1/networks?name=" + name, headers=headers)
    return r.json()["networks"][0]

def change_alias_ips(headers, server_id, payload):
    headers = headers.copy()
    headers["Content-Type"] = "application/json"
    r = requests.post("https://api.hetzner.cloud/v1/servers/{}/actions/change_alias_ips".format(server_id), data=json.dumps(payload), headers=headers)
    r.raise_for_status()

def assign_floating_ip(headers, floating_ip_id, payload):
    headers = headers.copy()
    headers["Content-Type"] = "application/json"
    r = requests.post("https://api.hetzner.cloud/v1/floating_ips/{}/actions/assign".format(floating_ip_id), data=json.dumps(payload), headers=headers)
    r.raise_for_status()

def add_ip(ip, interface):
    os.system("/bin/ip address add {}/32 dev {}".format(ip, interface))

def del_ip(ip, interface):
    os.system("/bin/ip address del {}/32 dev {}".format(ip, interface))

def main(arg_type, arg_name, arg_state):
    hostname = socket.gethostname()
    config_path = "/etc/hcloud-failover/{}.json".format(arg_name)
    log("Opening configuration {}".format(config_path))
    with open(config_path, "r") as config_file:
        config = json.load(config_file)
    headers = {
        "Authorization": "Bearer " + config["api_token"]
    }
    log("Perform action for transition to state {} for {}".format(arg_state, hostname))
    if arg_state == "MASTER":
        servers = get_servers(headers)
        if "floating_ips" in config and len(config["floating_ips"]) > 0:
            log("Configuring {} floating ips".format(len(config["floating_ips"])))
            server = next(x for x in servers if x["name"] == hostname)
            floating_ips = get_floating_ips(headers)
            for floating_ip_name in config["floating_ips"]:
                floating_ip = next(x for x in floating_ips if x["name"] == floating_ip_name)
                log("Adding floating ip {}".format(floating_ip))
                add_ip(floating_ip["ip"], config["floating_interface"])
                assign_floating_ip(headers, floating_ip["id"], {
                    "server": server["id"]
                })
                log("New master '{}' was assigned floating ip '{}' ({})".format(server["name"], floating_ip["name"], floating_ip["ip"]))
        if "alias_ips" in config and len(config["alias_ips"]) > 0:
            log("Configuring {} alias ips".format(len(config["alias_ips"])))
            network = get_network(headers, config["network"])
            for server in [x for x in servers if x["name"] in config["servers"]]:
                server_alias_ips = next(x["alias_ips"] for x in server["private_net"] if x["network"] == network["id"])
                count = 0
                for alias_ip in config["alias_ips"]:
                    if alias_ip in server_alias_ips:
                        server_alias_ips.remove(alias_ip)
                        count = count + 1
                if count > 0:
                    change_alias_ips(headers, server["id"], {
                        "network": network["id"],
                        "alias_ips": server_alias_ips
                    })
                    log("Failed master '{}' was unassigned {} alias ips on network '{}' - set aliases are {}".format(server["name"], count, config["network"], server_alias_ips))
                    break
            server = next(x for x in servers if x["name"] == hostname)
            server_alias_ips = next(x["alias_ips"] for x in server["private_net"] if x["network"] == network["id"])
            for alias_ip in config["alias_ips"]:
                server_alias_ips.append(alias_ip)
                log("Adding alias ip {}".format(alias_ip))
                add_ip(alias_ip, config["alias_interface"])
            change_alias_ips(headers, server["id"], {
                "network": network["id"],
                "alias_ips": server_alias_ips
            })
            log("New master '{}' was assigned alias ips {} on network '{}' - set aliases are: {}".format(server["name"], config["alias_ips"], config["network"], server_alias_ips))
    else:
        if "floating_ips" in config:
            for floating_ip in config["floating_ips"]:
                log("Deleting floating ip {}".format(floating_ip))
                del_ip(floating_ip, config["alias_interface"])
        if "alias_ips" in config:
            for alias_ip in config["alias_ips"]:
                log("Deleting alias ip {}".format(alias_ip))
                del_ip(alias_ip, config["alias_interface"])
    log("All done, exiting gracefully!")

def configure_keepalived_unicast_peer(arg_name):
    hostname = socket.gethostname()
    config_path = "/etc/hcloud-failover/{}.json".format(arg_name)
    log("Opening configuration {}".format(config_path))
    with open(config_path, "r") as config_file:
        config = json.load(config_file)
    headers = {
        "Authorization": "Bearer " + config["api_token"]
    }
    network = get_network(headers, config["network"])
    servers = get_servers(headers)
    other_ips = []
    for server in [x for x in servers if x["name"] != hostname and x["name"] in config["servers"]]:
        ip = next(x["ip"] for x in server["private_net"] if x["network"] == network["id"])
        other_ips.append(ip)
        log("Server {} was found to have IP {}".format(server["name"], ip))
    keepalived_config_path = "/etc/keepalived/conf/{}.conf".format(arg_name)
    with open(keepalived_config_path, "r") as keepalived_conf:
        conf = keepalived_conf.read()
    new_section = "\g<1>unicast_peer {" + os.linesep + os.linesep.join(["\g<1>\g<1>" + ip for ip in other_ips]) + os.linesep + "\g<1>}"
    new_conf = re.sub(r"^([^#]\s*)unicast_peer\s+{.*?}", new_section, conf, flags=re.MULTILINE|re.DOTALL)
    with open(keepalived_config_path, "w") as keepalived_conf:
        keepalived_conf.write(new_conf)

if __name__ == "__main__":
    syslog.openlog("hcloud-failover")
    if sys.argv[1] == "--configure-keepalived-unicast-peer":
        configure_keepalived_unicast_peer(arg_name=sys.argv[2])
    else:
        main(arg_type=sys.argv[1], arg_name=sys.argv[2], arg_state=sys.argv[3])
    syslog.closelog()