Как да направите свой собствен автоскалер за клъстер

Здравейте! Ние обучаваме хора за работа с големи данни. Невъзможно е да си представим образователна програма за големи данни без собствен клъстер, върху който всички участници работят заедно. Поради тази причина нашата програма винаги я има 🙂 Ние сме ангажирани с нейната конфигурация, настройка и администриране, а момчетата директно стартират MapReduce задания там и използват Spark.

В тази публикация ще ви разкажем как решихме проблема с неравномерното зареждане на клъстера, като написахме наш собствен автоматичен скалер с помощта на облака Облачни решения на Mail.ru.

проблем

Нашият клъстер не се използва в типичен режим. Изхвърлянето е силно неравномерно. Например, има практически занятия, когато всичките 30 души и един учител отиват в клъстера и започват да го използват. Или пак има дни преди крайния срок, когато натоварването се увеличава много. През останалото време клъстерът работи в режим на недотоварване.

Решение №1 е да запазите клъстер, който ще издържи на пиковите натоварвания, но ще бъде бездействащ през останалото време.

Решение №2 е да запазите малък клъстер, към който ръчно добавяте възли преди класове и по време на пикови натоварвания.

Решение #3 е да запазите малък клъстер и да напишете автоматичен скалер, който ще следи текущото натоварване на клъстера и, използвайки различни API, добавя и премахва възли от клъстера.

В тази публикация ще говорим за решение №3. Този автоскалер е силно зависим от външни фактори, а не от вътрешни, и доставчиците често не го предоставят. Ние използваме облачната инфраструктура на Mail.ru Cloud Solutions и написахме автоматичен скалер, използвайки MCS API. И тъй като ние учим как да работим с данни, решихме да покажем как можете да напишете подобен автоскалер за вашите собствени цели и да го използвате с вашия облак

Предпоставки

Първо, трябва да имате Hadoop клъстер. Например, използваме разпределението на HDP.

За да могат вашите възли да бъдат бързо добавяни и премахвани, трябва да имате определено разпределение на ролите между възлите.

1. Главен възел. Е, няма нужда да обяснявате нищо конкретно: основният възел на клъстера, на който например се стартира драйверът на Spark, ако използвате интерактивния режим.
2. Възел за дата. Това е възелът, на който съхранявате данни за HDFS и където се извършват изчисленията.
3. Изчислителен възел. Това е възел, където не съхранявате нищо в HDFS, но където се извършват изчисления.

Важен момент. Автоматичното мащабиране ще се случи поради възли от трети тип. Ако започнете да приемате и добавяте възли от втория тип, скоростта на реакция ще бъде много ниска - извеждането от експлоатация и повторното подаване ще отнеме часове на вашия клъстер. Това, разбира се, не е това, което очаквате от автоматичното мащабиране. Тоест не докосваме възли от първи и втори тип. Те ще представляват минимален жизнеспособен клъстер, който ще съществува по време на продължителността на програмата.

И така, нашият autoscaler е написан на Python 3, използва Ambari API за управление на клъстерни услуги, използва API от Mail.ru Cloud Solutions (MCS) за стартиране и спиране на машини.

Архитектура на решението

1. Модул autoscaler.py. Съдържа три класа: 1) функции за работа с Ambari, 2) функции за работа с MCS, 3) функции, свързани пряко с логиката на autoscaler.
2. Скрипт observer.py. По същество се състои от различни правила: кога и в какви моменти да се извикват функциите за автоматично мащабиране.
3. Конфигурационен файл config.py. Той съдържа например списък с възли, разрешени за автоматично мащабиране и други параметри, които влияят например на това колко време да се изчака от момента, в който е добавен нов възел. Има и времеви отметки за началото на класовете, така че преди класа да се стартира максимално разрешената конфигурация на клъстера.

Нека сега да разгледаме парчетата код в първите два файла.

1. Модул Autoscaler.py

Клас Амбари

Ето как изглежда парче код, съдържащ клас 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

По-горе, като пример, можете да разгледате изпълнението на функцията stop_all_services, което спира всички услуги на желания клъстерен възел.

На входа на класа Ambari преминаваш:

• ambari_url, например като 'http://localhost:8080/api/v1/clusters/',
• cluster_name – името на вашия клъстер в Ambari,
• headers = {'X-Requested-By': 'ambari'}
• и вътре auth ето вашето потребителско име и парола за Ambari: auth = ('login', 'password').

Самата функция не е нищо повече от няколко извиквания през REST API към Ambari. От логическа гледна точка първо получаваме списък с изпълнявани услуги на възел и след това искаме от даден клъстер, от даден възел, да прехвърлим услуги от списъка към състоянието INSTALLED. Функции за стартиране на всички услуги, за прехвърляне на възли в състояние Maintenance и т.н. изглеждат подобни - те са само няколко заявки през API.

Клас Mcs

Ето как изглежда парче код, съдържащ клас 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

На входа на класа Mcs предаваме идентификатора на проекта в облака и потребителския идентификатор, както и неговата парола. Във функция vm_turn_on искаме да включим една от машините. Логиката тук е малко по-сложна. В началото на кода се извикват три други функции: 1) трябва да получим токен, 2) трябва да преобразуваме името на хоста в името на машината в MCS, 3) да получим идентификатора на тази машина. След това просто правим заявка за публикуване и стартираме тази машина.

Ето как изглежда функцията за получаване на токен:

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

Клас на Autoscaler

Този клас съдържа функции, свързани със самата операционна логика.

Ето как изглежда част от кода за този клас:

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

Приемаме класове за влизане. Ambari и Mcs, списък с възли, които са разрешени за мащабиране, както и параметри за конфигуриране на възел: памет и процесор, разпределени за възела в YARN. Има и 2 вътрешни параметъра q_ram, q_cpu, които са опашки. Използвайки ги, ние съхраняваме стойностите на текущото натоварване на клъстера. Ако видим, че през последните 5 минути е имало постоянно увеличено натоварване, тогава решаваме, че трябва да добавим +1 възел към клъстера. Същото важи и за състоянието на недостатъчно използване на клъстера.

Кодът по-горе е пример за функция, която премахва машина от клъстера и я спира в облака. Първо има извеждане от експлоатация YARN Nodemanager, след което режимът се включва Maintenance, след което спираме всички услуги на машината и изключваме виртуалната машина в облака.

2. Скрипт наблюдател.py

Примерен код от там:

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} has been successfully scaled-up".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)

В него проверяваме дали са създадени условия за увеличаване на капацитета на клъстера и дали има резервни машини, получаваме името на хоста на една от тях, добавяме я към клъстера и публикуваме съобщение за това в Slack на нашия екип. След което започва cooldown_period, когато не добавяме или премахваме нищо от клъстера, а просто наблюдаваме натоварването. Ако се е стабилизирал и е в коридора на оптималните стойности на натоварване, тогава просто продължаваме наблюдението. Ако един възел не е достатъчен, тогава добавяме още един.

За случаите, когато ни предстои урок, вече знаем със сигурност, че един възел няма да е достатъчен, така че веднага стартираме всички свободни възли и ги поддържаме активни до края на урока. Това се случва с помощта на списък с времеви клеймца на дейността.

Заключение

Autoscaler е добро и удобно решение за случаите, когато имате неравномерно натоварване на клъстера. Едновременно постигате желаната конфигурация на клъстера за пикови натоварвания и в същото време не задържате този клъстер по време на недостатъчно натоварване, спестявайки пари. Е, освен това всичко това се случва автоматично, без вашето участие. Самият autoscaler не е нищо повече от набор от заявки към API на мениджъра на клъстера и API на облачния доставчик, написани според определена логика. Това, което определено трябва да запомните, е разделянето на възлите на 3 вида, както писахме по-рано. И ще бъдете щастливи.

Източник: www.habr.com