🥇Jak vytvořit svůj autoscaler pro cluster

Ahoj! Vzděláváme lidi v práci s velkými daty. Je těžké si představit vzdělávací program o velkých datech bez vlastního clustru, na kterém všichni účastníci společně pracují. Z tohoto důvodu je na našem programu vždy k dispozici 🙂 Zajišťujeme jeho nastavení, ladění a administraci, a naši studenti tam přímo spouští MapReduce úlohy a využívají Spark.

V tomto příspěvku vám povíme, jak jsme vyřešili problém s nerovnoměrným zatížením clustru tím, že jsme napsali vlastní autoskalér pomocí cloudu. Mail.ru Cloud Solutions.

Problém

Náš klastr se nepoužívá zcela obvyklým způsobem. Utilizace je velmi nerovnoměrná. Například jsou praktické cvičení, kdy se na klastru sejde všech 30 lidí a lektor, a začínají ho používat. Nebo jsou dny před termínem, kdy zatížení výrazně vzrůstá. V ostatních časech klastr funguje v režimu nedostatečného zatížení.

Řešení č. 1 – udržet klastr, který zvládne špičkové zatížení, ale jinak bude nečinný.

Řešení č. 2 – mít malý klastr, do kterého ručně přidávat uzly před cvičeními a během špičkového zatížení.

Řešení č. 3 – mít malý klastr a napsat autoskalér, který bude sledovat aktuální zatížení clustru a automaticky, pomocí různých API, přidávat a odstraňovat uzly z clustru.

V tomto příspěvku se budeme zabývat řešením č. 3. Takový autoskalér silně závisí na vnějších faktorech, nikoli na vnitřních, a poskytovatelé ho často nenabízejí. My využíváme cloudovou infrastrukturu Mail.ru Cloud Solutions a napsali jsme autoskalér pomocí API MCS. A protože vzděláváme v oblasti dat, rozhodli jsme se ukázat, jak si můžete podobný autoskalér napsat pro své účely a používat ho se svým cloudem.

Požadavky

Za prvé, musíte mít Hadoop klastr. My například používáme distribuci HDP.

Aby mohly být uzly rychle přidávány a odstraňovány, musíte mít určitý rozpis rolí mezi uzly.

Hlavní uzel. Tady není potřeba nic víc vysvětlovat: hlavní uzel clustru, na kterém se spouští například řidič Spark, pokud používáte interaktivní režim.
Data uzel. To je uzel, na kterém jsou uložena data v HDFS a na kterém probíhají výpočty.
Výpočetní uzel. Je to uzel, na kterém se nic neukládá na HDFS, ale na něm probíhají výpočty.

Důležitý bod. Automatické škálování bude probíhat na základě třetího typu uzlů. Pokud začnete odebírat a přidávat uzly druhého typu, rychlost reakce bude výrazně nízká – dekomise a rekomise budou trvat hodiny na vašem clusteru. To rozhodně není to, co očekáváte od automatického škálování. Tedy uzly prvního a druhého typu se nedotýkáme. Budou představovat minimálně životaschopný cluster, který bude existovat po celou dobu programu.

Náš automatický škálovač je napsán v Pythonu 3, používá API Ambari k řízení služeb clusteru, využívá API od Mail.ru Cloud Solutions (MCS) pro spuštění a zastavení strojů.

Architektura řešení

Modul autoscaler.py. Obsahuje tři třídy: 1) funkce pro práci s Ambari, 2) funkce pro práci s MCS, 3) funkce, které se přímo týkají logiky fungování automatického škálovače.
Skript observer.py. V podstatě se skládá z různých pravidel: kdy a v jakých momentech volat funkce automatického škálovače.
Soubor s konfiguračními parametry config.py. Obsahuje například seznam uzlů povolených pro automatické škálování a další parametry, které ovlivňují to, jak dlouho čekat od chvíle, kdy byl přidán nový uzel. Také obsahuje časové značky začátků sezení, aby byla před sezením spuštěna maximálně povolená konfigurace clusteru.

Nyní se podívejme na úryvky kódu nacházející se uvnitř prvních dvou souborů.

1. Modul autoscaler.py

Třída Ambari

Takto vypadá úryvek kódu obsahující třídu Ambari:

třída Ambari:
    def __init__(self, ambari_url, cluster_name, headers, auth):
        self.ambari_url = ambari_url
        self.cluster_name = cluster_name
        self.headers = headers
        self.auth = auth

    def stop_all_services(self, hostname):
        url = self.ambari_url + self.cluster_name + ' /hosts/ ' + hostname + ' /host_components /'
        url2 = self.ambari_url + self.cluster_name + ' /hosts/ ' + hostname
        req0 = requests.get(url2, headers=self.headers, auth=self.auth)
        services = req0.json()['host_components']
        services_list = list(map(lambda x: x['HostRoles']['component_name'], services))
        data = {
            "RequestInfo": {
                "context":"Stop All Host Components",
                "operation_level": {
                    "level":"HOST",
                    "cluster_name": self.cluster_name,
                    "host_names": hostname
                },
                "query":"HostRoles /component_name.in({0})".format(",".join(services_list))
            },
            "Body": {
                "HostRoles": {
                    "state":"INSTALLED"
                }
            }
        }
        req = requests.put(url, data=json.dumps(data), headers=self.headers, auth=self.auth)
        if req.status_code in [200, 201, 202]:
            message = 'Žádost byla přijata'
        else:
            message = req.status_code
        return message

Výše uvedené jako příklad lze vidět implementaci funkce stop_all_services, která zastavuje všechny služby na požadované uzlu clusteru.

Do třídy Ambari předáváte:

ambari_url, například ve formátu 'http://localhost:8080/api/v1/clusters/',
cluster_name – název vašeho clusteru v Ambari,
headers = {'X-Requested-By': 'ambari'}
a uvnitř auth se nachází váš login a heslo od Ambari: auth = ('login', 'password').

Funkce sama o sobě představuje maximálně pár volání pomocí REST API k Ambari. Z hlediska logiky nejprve získáme seznam běžících služeb na uzlu a poté žádáme na tomto clusteru, na tomto uzlu převést služby ze seznamu do stavu INSTALLED. Funkce pro spuštění všech služeb, pro převod uzlů do stavu Maintenance a další se podobají – to jsou prostě několik žádostí přes API.

třída Mcs

Takto vypadá úryvek kódu obsahující třídu Mcs:

třída Mcs:
    def __init__(self, id1, id2, password):
        self.id1 = id1
        self.id2 = id2
        self.password = password
        self.mcs_host = 'https://infra.mail.ru:8774/v2.1'

    def vm_turn_on(self, hostname):
        self.token = self.get_mcs_token()
        host = self.hostname_to_vmname(hostname)
        vm_id = self.get_vm_id(host)
        mcs_url1 = self.mcs_host + ' /servers /' + self.vm_id + ' /action'
        headers = {
            'X-Auth-Token': '{0}'.format(self.token),
            'Content-Type': 'application /json'
        }
        data = {'os-start' : 'null'}
        mcs = requests.post(mcs_url1, data=json.dumps(data), headers=headers)
        return mcs.status_code

Do třídy Mcs předáváme id projektu uvnitř cloudu a id uživatele, stejně jako jeho heslo. V funkci vm_turn_on Chceme zapnout jeden z serverů. Logika je zde o něco složitější. Na začátku kódu se volají tři další funkce: 1) potřebujeme získat token, 2) musíme převést hostname na název serveru v MCS, 3) získat ID tohoto serveru. Dále provedeme prostý POST požadavek a spustíme tento server.

Takto vypadá samotná funkce pro získání tokenu:

def get_mcs_token(self):
        url = 'https://infra.mail.ru:35357/v3/auth/tokens?nocatalog'
        headers = {'Content-Type': 'application/json'}
        data = {
            'auth': {
                'identity': {
                    'methods': ['password'],
                    'password': {
                        'user': {
                            'id': self.id1,
                            'password': self.password
                        }
                    }
                },
                'scope': {
                    'project': {
                        'id': self.id2
                    }
                }
            }
        }
        params = (('nocatalog', ''),)
        req = requests.post(url, data=json.dumps(data), headers=headers, params=params)
        self.token = req.headers['X-Subject-Token']
        return self.token

Třída Autoscaler

Tato třída obsahuje funkce týkající se samotné logiky fungování.

Takto vypadá část kódu této třídy:

třída Autoscaler:
    def __init__(self, ambari, mcs, scaling_hosts, yarn_ram_per_node, yarn_cpu_per_node):
        self.scaling_hosts = scaling_hosts
        self.ambari = ambari
        self.mcs = mcs
        self.q_ram = deque()
        self.q_cpu = deque()
        self.num = 0
        self.yarn_ram_per_node = yarn_ram_per_node
        self.yarn_cpu_per_node = yarn_cpu_per_node

    def scale_down(self, hostname):
        flag1 = flag2 = flag3 = flag4 = flag5 = False
        if hostname in self.scaling_hosts:
            while True:
                time.sleep(5)
                status1 = self.ambari.decommission_nodemanager(hostname)
                if status1 == 'Request accepted' or status1 == 500:
                    flag1 = True
                    logging.info('Žádost o dekomisování byla přijata: {0}'.format(flag1))
                    break
            while True:
                time.sleep(5)
                status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
                if status3 == 'INSTALLED':
                    flag3 = True
                    logging.info('Nodemaneger byl dekomisován: {0}'.format(flag3))
                    break
            while True:
                time.sleep(5)
                status2 = self.ambari.maintenance_on(hostname)
                if status2 == 'Request accepted' or status2 == 500:
                    flag2 = True
                    logging.info('Žádost o údržbu byla přijata: {0}'.format(flag2))
                    break
            while True:
                time.sleep(5)
                status4 = self.ambari.check_maintenance(hostname, 'NODEMANAGER')
                if status4 == 'ON' or status4 == 'IMPLIED_FROM_HOST':
                    flag4 = True
                    self.ambari.stop_all_services(hostname)
                    logging.info('Údržba je zapnuta: {0}'.format(flag4))
                    logging.info('Zastavování služeb')
                    break
            time.sleep(90)
            status5 = self.mcs.vm_turn_off(hostname)
            while True:
                time.sleep(5)
                status5 = self.mcs.get_vm_info(hostname)['server']['status']
                if status5 == 'SHUTOFF':
                    flag5 = True
                    logging.info('VM je vypnuta: {0}'.format(flag5))
                    break
            if flag1 and flag2 and flag3 and flag4 and flag5:
                message = 'Úspěch'
                logging.info('Snížení proběhlo úspěšně')
                logging.info('Začala doba chlazení. Počkejte několik minut')
        return message

Na vstupu přijímáme třídy Ambari a Mcs, seznam uzlů, které jsou povoleny pro škálování, a také parametry konfigurace uzlů: paměť a CPU, přidělené na uzel v YARN. Také máme 2 interní parametry q_ram, q_cpu, které jsou frontami. Pomocí nich uchováváme hodnoty aktuální zátěže clusteru. Pokud vidíme, že v posledních 5 minutách byla trvale zvýšená zátěž, rozhodujeme se přidat +1 uzel do clusteru. Totéž platí pro stav podzátěže clusteru.

V kódu výše je uveden příklad funkce, která odstraňuje stroj z clusteru a zastavuje ho v cloudu. Nejprve dochází k dekomisování YARN Nodemanager, dále se aktivuje režim Maintenance, poté zastavujeme všechny služby na stroji a vypínáme virtuální stroj v cloudu.

2. Skript observer.py

Příklad kódu z tamtud:

if scaler.assert_up(config.scale_up_thresholds) == True:
        hostname = cloud.get_vm_to_up(config.scaling_hosts)
        if hostname != None:
            status1 = scaler.scale_up(hostname)
            if status1 == 'Success':
                text = {"text": "{0} byl úspěšně zvětšen".format(hostname)}
                post = {"text": "{0}".format(text)}
                json_data = json.dumps(post)
                req = requests.post(webhook, data=json_data.encode('ascii'), headers={'Content-Type': 'application/json'})
                time.sleep(config.cooldown_period*60)

V něm kontrolujeme, zda nastaly podmínky pro zvýšení kapacity clustera a zda jsou k dispozici volné stroje, získáváme hostname jednoho z nich, přidáváme ho do clustera a zveřejňujeme o tom zprávu na Slacku našeho týmu. Poté se spustí cooldown_period, kdy nic nepřidáváme ani neodstraňujeme z clustera, ale pouze monitorujeme zatížení. Pokud se stabilizovalo a nachází se v optimálním rozmezí zatížení, pokračujeme v monitorování. Pokud však jedna uzel nestačila, přidáme další.

V případech, kdy nás čeká vyučování, již víme jistě, že jedna uzel nestačí, proto ihned spouštíme všechny dostupné uzly a držíme je aktivní až do konce vyučování. To se děje pomocí seznamu časových značek vyučování.

Závěr

Autoscaler je dobré a pohodlné řešení pro případy, kdy máte nerovnoměrné zatížení clustera. Zároveň dosahujete potřebné konfigurace clustera pro špičkové zatížení a přitom si tento cluster během podzátěže neudržujete, čímž šetříte prostředky. A navíc se to vše děje automatizovaně bez vašeho zásahu. Samotný autoscaler není nic víc než soubor dotazů k API správce clustera a API poskytovatele cloudu, napsaných podle určité logiky. Na co je třeba nezapomínat, je rozdělení uzlů do 3 typů, jak jsme psali dříve. A budete mít štěstí.

Zdroj: habr.com

Jak vytvořit vlastní autoskalér pro klastr