🥇Kuidas teha oma autoskaleerija klastrile

Tere! Me õpetame inimesi töötama suurte andmetega. Haridusprogrammi, mis käsitleb suurte andmete teemade õpetamist, ei saa ette kujutada ilma oma klastrita, kus kõik osalejad suudavad koos töötada. Sellepärast on meie programmis see alati olemas 🙂 Me tegeleme selle seadistamise, häälestamise ja haldamisega, samas kui osalejad käivitavad seal MapReduce-töid ja kasutavad Spark'i.

Selles postituses räägime, kuidas me lahendasime klastri ebavõrdse koormuse probleemi, kirjutades oma automaatse skaaleerija, kasutades pilveteenust. Mail.ru Pilve Lahendused.

Probleem

Meie klastrit kasutatakse mitte just tüüpilises režiimis. Kasutamine on väga ebavõrdne. Näiteks on praktikumipäevad, mil kõik 30 inimest ja õpetaja siseneb klastrisse ja hakkavad seda kasutama. Või on ka päevi enne tähtaega, mil koormus tõuseb märgatavalt. Kõigil teistel aegadel töötab klaster alakoormuse režiimis.

Lahendus nr 1 on hoida klaster, mis suudab taluda tipukoormusi, kuid on muul ajal seisakutos.

Lahendus nr 2 on hoida väikest klastrit, kuhu lisada node’id käsitsi enne tunde ja tipukoormuste ajal.

Lahendus nr 3 on hoida väikest klastrit ja kirjutada automaatne skaaleerija, mis jälgib klastri hetkekoormust ja automaatselt, kasutades erinevaid API-sid, lisab ja eemaldab node’e klastri seest.

Selles postituses räägime lahendusest nr 3. Selline automaatne skaaleerija sõltub tugevalt välistest teguritest, mitte sisemistest, ja teenusepakkujad ei paku seda sageli. Me kasutame Mail.ru Cloud Solutions’i pilveuuringute infrastruktuuri ja oleme kirjutanud automaatse skaaleerija, kasutades MCS API-d. Kuna me õpetame andmete töötlemist, otsustasime näidata, kuidas saate kirjutada sellise automaatse skaaleerija oma eesmärkide täitmiseks ja kasutada seda oma pilves.

Eeltingimused

Esiteks peab teil olema Hadoopi klaster. Me kasutame näiteks HDP distributsiooni.

Et node’id saaksid kiiresti lisanduda ja eemalduda, peab teil olema määratud rollijagamine node'ite vahel.

Peamoodul. Siin pole erilist seletamist vaja: peamine node klastris, kus käivitatakse näiteks Spark’i draiver, kui kasutate interaktiivset režiimi.
Andmoodul. See on modul, kus hoitakse andmeid HDFS-is ja kus toimub ka arvutused.
Arvutusnode. See node, kus teil ei ole midagi HDFS-is salvestatud, kuid kus toimub arvutustöö.

Oluline punkt. Automaatne skalaarimine toimub kolmanda tüübi node'ide arvelt. Kui hakkate lisama ja eemaldama teise tüübi node'e, siis reageerimiskiirus langeb oluliselt – dekomeerimine ja rekomeerimine võtab teie klastris tundide viisi. See ei ole see, mida ootate automaatse skalaarimise puhul. Seega esimest ja teist tüüpi node'e me ei muuda. Need moodustavad minimaalsete elujõuliste node'ide k Lastri, mis eksisteerib programmi kestuse vältel.

Nii et meie automaatne skaaler on kirjutatud Python 3-s, kasutades Ambari API-d klastriteenuste haldamiseks, kasutab Mail.ru Cloud Solutions API-d (MCS) masinate käivitamiseks ja peatamiseks.

Lahenduse arhitektuur

Moodul autoscaler.py. Selles on määratletud kolm klassi: 1) funktsioonid Ambari tööks, 2) funktsioonid MCS tööks, 3) funktsioonid, mis on otseselt seotud automaatse skaaleri tööloogikaga.
Skript observer.py. See koosneb erinevatest reeglitest: millal ja millistel hetkedel kutsuda automaatse skaaleri funktsioone.
Configuratsiooniparameetrite fail config.py. Seal on näiteks loetelu node'idest, mis on lubatud automaatseks skaalamiseks, ja muud parameetrid, mis mõjutavad näiteks seda, kui kaua oodata, alates hetkest, mil uus node lisati. Samuti on seal algusajad, et enne tegevust oleks käivitatud maksimaalselt lubatud klastrikonfiguratsioon.

Vaadakem nüüd koodijuppe, mis asuvad esimeses kahes failis.

1. Moodul autoscaler.py

Ambari klass

Näin välja osa koodist, mis sisaldab klassi Ambari:

class 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 = 'Request accepted'
        else:
            message = req.status_code
        return message

Ülal toodud näites on näha funktsiooni rakendust stop_all_services, mis peatab kõik teenused soovitud klastrisõlmes.

Klassi sisendiks Ambari antakse:

ambari_url, näiteks järgmisega 'http:\/\/localhost:8080\/api\/v1\/clusters\/ ',
cluster_name – teie klastrinimi Ambaris,
headers = {'X-Requested-By': 'ambari'}
ja sees on auth teie Ambari kasutajanimi ja parool: auth = ('login', 'password').

Funktsioon koosneb mitte rohkem kui paarist pöördumisest Ambari REST API kaudu. Loogika poolelt saame esmalt nimekirja rakendatud teenustest sõlmes ja palume siis katta need klastris ja antud sõlmes olekusse INSTALLED. Kõik teenuste käivitamise funktsioonid, sõlmede viimine seisundisse Maintenance ja muud sarnane näevad välja sarnasti – need on lihtsalt mõned päringud API kaudu.

Klass Mcs

Näin välja osa koodist, mis sisaldab klassi Mcs:

class 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

Klassi sisendiks Mcs me anname projektide id pilves ja kasutaja id, samuti tema parooli. Funktsioonis vm_turn_on Me tahame sisse lülitada ühe masinast. Siin on veidi keerulisem loogika. Koodi alguses kutsutakse välja kolm muud funktsiooni: 1) meil on vaja saada token, 2) meil on vaja konverteerida hostname masina nimeks MCS-is, 3) saada selle masina id. Seejärel teeme lihtsalt postipära ja käivitame selle masina.

Siin on funktsioon tokeni saamiseks:

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

Klass Autoscaler

Selles klassis on funktsioonid, mis on seotud loogikaga.

Siin on selle klassi koodilõik:

class 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('Decomission request accepted: {0}'.format(flag1))
                    break
            while True:
                time.sleep(5)
                status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
                if status3 == 'INSTALLED':
                    flag3 = True
                    logging.info('Nodemaneger decommissioned: {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('Maintenance request accepted: {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('Maintenance is on: {0}'.format(flag4))
                    logging.info('Stopping services')
                    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 is turned off: {0}'.format(flag5))
                    break
            if flag1 and flag2 and flag3 and flag4 and flag5:
                message = 'Success'
                logging.info('Scale-down finished')
                logging.info('Cooldown period has started. Wait for several minutes')
        return message

Sisendiks võtame klassid Ambari ja Mcs, skaleerimiseks lubatud sõlmede loendi, samuti sõlme konfiguratsiooni parameetrid: YARN-is eraldatud mälumaht ja CPU. Samuti on meil kaks sisemist parameetrit q_ram, q_cpu, mis on järjekorrad. Nende abil salvestame klastrite praeguste koormuse väärtusi. Kui näeme, et viimase viie minuti jooksul on koormus pidevalt kõrge, siis otsustame, et on vajalik klastrisse lisada +1 sõlm. Sama kehtib ka klastrite alakoormuse olukorra kohta.

Ülaltoodud koodis on näide funktsioonist, mis eemaldab masina klastrist ja peatab selle pilves. Alguses toimub dekomeerimine YARN Nodemanager, siis lülitame sisse režiimi Maintenance, siis peatame kõik teenused masinas ja lülitame välja virtuaalse masina pilves.

2. Skript observer.py

Koodinäide sealt:

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} on edukalt suurendatud".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)

Siin kontrollime, kas klastrite võimsuste suurendamise tingimused on tekkinud ja kas varuks on masinad, saame ühe nende hostnimedest, lisame selle klastrisse ja avaldame sellest meie meeskonna Slackis teadet. Pärast seda algab cooldown_period, mil me ei lisa ega eemalda midagi klastrist, vaid jälgime lihtsalt koormust. Kui see stabiliseerub ja jääb optimaalse koormuse väärtuste vahemikku, jätkame lihtsalt jälgimist. Kui aga üks node ei piisanud, siis lisame veel ühe.

Olukordade puhul, kui meil on ees tegevus, teame juba kindlasti, et üks node ei piisa, seetõttu käivitame kohe kõik vabad node'id ja hoiame need aktiivsena kuni tegevuse lõpuni. See toimub ajastamiste nimekirja abil.

Kokkuvõte

Autoskaleerija on hea ja mugav lahendus, kui teie klastris esineb ebaühtlane koormus. Saate samal ajal saavutada vajaliku klastrikonfiguratsiooni tippkoormuste jaoks ning säästa raha, hoides klastrit tühja ajal. Ja kõik see juhtub automaatselt teie osaluseta. Ise autoskaleerija ei ole midagi muud kui korduste komplekt API päringutest klaster-halduriga ja pilveteenuse pakkujaga, mis on kirjutatud teatud loogika järgi. Mida tuleks kindlasti meeles pidada, on node'ide jagamine 3 tüüpi, nagu me varem kirjutasime. Ja see toob teile õnne.

Allikas: habr.com

Kuidas luua oma auto skaalautuja klastrile