🥇Sådan laver du din egen auto-skalering for klusteret

Hej! Vi uddanner folk i at arbejde med store data. Det er umuligt at forestille sig et uddannelsesprogram for store data uden en klynge, hvor alle deltagere arbejder sammen. Af denne grund er den altid til stede i vores program 🙂 Vi håndterer dens opsætning, tuning og administration, mens deltagerne kører MapReduce-jobbene og bruger Spark.

I dette indlæg vil vi fortælle, hvordan vi løste problemet med uensartet belastning af klyngen ved at skrive vores egen autoscaler ved hjælp af skyen. Mail.ru Cloud Solutions.

Problemet

Klyngen bruges ikke helt i et typisk mode. Udtjening er meget ujævn. For eksempel er der praktiske sessioner, hvor alle 30 personer og instruktøren går ind i klyngen og begynder at bruge den. Eller igen er der dage før deadline, hvor belastningen stiger kraftigt. I al anden tid fungerer klyngen i en tilstand af underbelastning.

Løsning nr. 1 er at have en klynge, der kan håndtere spidsbelastninger, men som vil stå stille i al anden tid.

Løsning nr. 2 er at have en lille klynge, hvor man manuelt tilføjer noder før lektioner og under spidsbelastning.

Løsning nr. 3 er at have en lille klynge og skrive en autoscaler, der overvåger den aktuelle belastning af klyngen og automatisk, ved hjælp af forskellige API'er, tilføjer og fjerner noder fra klyngen.

I dette indlæg vil vi tale om løsning nr. 3. Sådan en autoscaler afhænger stærkt af eksterne faktorer snarere end interne, og udbydere tilbyder ofte ikke dette. Vi bruger Mail.ru Cloud Solutions' cloud-infrastruktur og har skrevet en autoscaler ved hjælp af MCS API. Da vi uddanner folk i datahåndtering, har vi besluttet at vise, hvordan du kan skrive en lignende autoscaler til dine formål og bruge den med din egen cloud.

Forudsætninger

For det første skal du have et Hadoop-klynge. For eksempel bruger vi distributionen HDP.

For at noderne kan tilføjes og fjernes hurtigt, skal du have en bestemt rollefordeling på noderne.

Master-node. Her er der ikke meget at forklare: hovednoden i klyngen, hvor for eksempel Spark-driveren kører, hvis du bruger interaktiv tilstand.
Data-node. Dette er noden, hvor dine data opbevares på HDFS, og hvor beregningerne også finder sted.
Beregningsnode. Dette er en node, hvor du ikke opbevarer noget på HDFS, men hvor beregninger finder sted.

Vigtig pointe. Autoskalering vil ske ved hjælp af noder af den tredje type. Hvis du begynder at tage noder af den anden type væk og tilføje dem, vil responstiden blive betydeligt lavere - dekommissionering og rekmissionering vil tage timer på dit cluster. Dette er selvfølgelig ikke, hvad du forventer af autoskalering. Så noder af den første og anden type tager vi ikke fat på. De vil udgøre et minimalt levedygtigt cluster, der vil eksistere i hele programmets varighed.

Så vores autoskalør er skrevet i Python 3, bruger Ambari API til at administrere cluster-tjenester, bruger API fra Mail.ru Cloud Solutions (MCS) til at starte og stoppe maskiner.

Løsningsarkitektur

Module autoscaler.py. Den omfatter tre klasser: 1) funktioner til at arbejde med Ambari, 2) funktioner til at arbejde med MCS, 3) funktioner, der vedrører selve logikken for autoskaløren.
Script observer.py. Den består hovedsageligt af forskellige regler: hvornår og på hvilke tidspunkter funktionerne for autoskaløren skal kaldes.
Filen med konfigurationsparametre config.py. Den indeholder eksempelvis en liste over noder, der er tilladt for autoskalering, samt andre parametre, der påvirker, hvor længe man skal vente fra det øjeblik, en ny node blev tilføjet. Her findes også tidsstempler for begyndelsen af lektionerne, så maksimalt tilladt cluster-konfiguration er startet før lektionen.

Lad os nu se på koden i de første to filer.

1. Modulet autoscaler.py

Klasse Ambari

Sådan ser et stykke kode ud, der indeholder klassen 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

Ovenfor kan man se implementeringen af funktionen stop_all_services, som stopper alle tjenester på den ønskede node i klyngen.

Til klassen Ambari giver du følgende input:

ambari_url, for eksempel af typen 'http://localhost:8080/api/v1/clusters/',
cluster_name – navnet på din klynge i Ambari,
headers = {'X-Requested-By': 'ambari'}
og inde i auth finder du dit brugernavn og kodeord til Ambari: auth = ('login', 'password').

Selve funktionen er ikke mere end et par kald via REST API til Ambari. Logisk set får vi først listen over kørende tjenester på noden, og derefter beder vi om at ændre disse tjenester i klyngen og på noden til status INSTALLED. Funktionerne til at starte alle tjenesterne, ændre noder til status Maintenance og så videre ser ens ud – det er blot flere API-kald.

Klassen Mcs

Sådan ser et stykke kode ud, der indeholder klassen 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

Til klassen Mcs vi sender projekt-id'et inden for skyen og bruger-id'et samt hans adgangskode. I funktionen vm_turn_on Vi ønsker at tænde en af maskinerne. Logikken her er lidt mere kompleks. I begyndelsen af koden kaldes tre andre funktioner: 1) vi skal have en token, 2) vi skal konvertere hostname til maskinnavnet i MCS, 3) få id'et for denne maskine. Derefter laver vi blot en post-anmodning og starter denne maskine.

Sådan ser selve funktionen til at få token ud:

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

Klassen Autoscaler

I denne klasse findes funktioner, der relaterer sig til selve logikken i arbejdet.

Sådan ser en del af koden fra denne klasse ud:

klasse 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('Dekommissioneringsanmodning accepteret: {0}'.format(flag1))
                    break
            while True:
                time.sleep(5)
                status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
                if status3 == 'INSTALLED':
                    flag3 = True
                    logging.info('Nodemanager dekommissioneret: {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('Vedligeholdelsesanmodning accepteret: {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('Vedligeholdelse er tændt: {0}'.format(flag4))
                    logging.info('Stopper tjenester')
                    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 er slukket: {0}'.format(flag5))
                    break
            if flag1 and flag2 and flag3 and flag4 and flag5:
                message = 'Succes'
                logging.info('Skalering ned færdig')
                logging.info('Cooldown-periode er startet. Vent et par minutter')
        return message

Vi modtager klasser Ambari og Mcs, en liste over noder, der er tilladt til skalering, samt konfigurationsparametre for noder: hukommelse og cpu, der er tildelt til noden i YARN. Derudover er der 2 interne parametre q_ram, q_cpu, som er køer. Ved hjælp af dem gemmer vi værdierne for den aktuelle belastning i klyngen. Hvis vi ser, at der har været en stabil høj belastning i de sidste 5 minutter, beslutter vi at tilføje +1 node til klyngen. Det samme gælder for tilstanden for overbelastning i klyngen.

I koden ovenfor er der et eksempel på en funktion, der fjerner en maskine fra klyngen og slukker den i skyen. Først sker der en dekommissionering YARN Nodemanager, så aktiveres tilstand Maintenance, og derefter stopper vi alle tjenester på maskinen og slukker den virtuelle maskine i skyen.

2. Script observer.py

Eksempel på kode derfra:

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} er blevet skaleret op med succes".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)

Her tjekker vi, om betingelserne for at øge klusterkapaciteten er opfyldt, og om der overhovedet er maskiner i reserve, får navnet på en af dem, tilføjer den til klusteret og offentliggør en meddelelse om dette i vores teams Slack. Derefter starter vi cooldown_period, hvor vi ikke tilføjer eller fjerner noget fra klusteret, men blot overvåger belastningen. Hvis den er stabil og ligger inden for det optimale belastningsinterval, fortsætter vi blot overvågningen. Hvis én node ikke var nok, tilføjer vi en til.

I tilfælde hvor vi har en aktivitet foran os, ved vi allerede med sikkerhed, at én node ikke vil være tilstrækkelig, så vi starter straks alle tilgængelige noder og holder dem aktive til aktiviteten er slut. Dette sker ved hjælp af en liste over tidsstempler for aktiviteter.

Konklusion

Auto-skaleren er en god og praktisk løsning til de tilfælde, hvor du oplever ujævn belastning på klusteret. Du opnår samtidig den ønskede konfiguration af klusteret til spidsbelastninger og holder ikke klusteret aktivt under lav belastning, hvilket sparer penge. Og det hele foregår automatiseret uden din deltagelse. Selve auto-skaleren er ikke mere end et sæt anmodninger til API'et for klusteradministratoren og API'et for den cloud-udbyder, der er skrevet efter en bestemt logik. Det, du skal huske på, er opdelingen af noderne i 3 typer, som vi tidligere har skrevet om. Og så vil lykken være din.

Kilde: habr.com