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

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

В този пост ще ви разкажем как решихме проблема с неравномерното натоварване на кластера, създавайки свой автоскейлер, използвайки облака Mail.ru Cloud Solutions.

Проблема

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

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

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

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

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

Предварителни условия

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

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

Майстор възел. Тук не е нужно много пояснения: основният възел на кластера, на който се стартира, например, драйвера на Spark, ако ползвате интерактивен режим.
Дата-нода. Това е нода, на която се съхраняват данните в HDFS и на която се извършват изчисления.
Изчислителна нода. Това е нода, на която не се съхраняват данни в HDFS, но се извършват изчисления.

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

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

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

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

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

1. Модул autoscaler.py

Клас Ambari

Ето как изглежда част от кода, съдържаща класа 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":"Спиране на всички хост компоненти",
                "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 = 'Заявката е приета'
        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 предиаваме id на проекта в облака и id на потребителя, както и паролата му. В функцията vm_turn_on Искаме да включим една от машините. Логиката тук е малко по-сложна. В началото на кода се извикват три други функции: 1) трябва да получим токен, 2) трябва да конвертираме името на хоста в името на машината в MCS, 3) да получим ID на тази машина. След това просто правим POST заявка и стартираме тази машина.

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

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('Заявка за декомисия приета: {0}'.format(flag1))
                    break
            while True:
                time.sleep(5)
                status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
                if status3 == 'INSTALLED':
                    flag3 = True
                    logging.info('Nodemaneger декомисия: {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('Заявка за поддръжка приета: {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('Поддръжката е активна: {0}'.format(flag4))
                    logging.info('Спиране на услугите')
                    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 е изключен: {0}'.format(flag5))
                    break
            if flag1 and flag2 and flag3 and flag4 and flag5:
                message = 'Успех'
                logging.info('Степента на намаляване завърши')
                logging.info('Период на охлаждане започна. Изчакайте няколко минути')
        return message

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

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

2. Скрипт observer.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} беше успешно увеличен".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, когато не добавяме и не премахваме нищо от клъстера, а просто наблюдаваме натоварването. Ако се стабилизира и е в коридора на оптимални натоварвания, просто продължаваме наблюдението. Ако обаче една нода не е достатъчна, добавяме още една.

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

Заключение

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

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