🥇როგორ გააკეთოთ თქვენი საკუთარი ავტოსკალერი კლასტერისთვის

გამარჯობა! ჩვენ ვამზადებთ ადამიანებს დიდ მონაცემებთან მუშაობისთვის. შეუძლებელია წარმოვიდგინოთ საგანმანათლებლო პროგრამა დიდ მონაცემებზე საკუთარი კლასტერის გარეშე, რომელზეც ყველა მონაწილე ერთად მუშაობს. ამ მიზეზით, ჩვენს პროგრამას ყოველთვის აქვს ის 🙂 ჩვენ ჩართული ვართ მის კონფიგურაციაში, ტუნინგსა და ადმინისტრირებაში და ბიჭები პირდაპირ ამუშავებენ MapReduce სამუშაოებს იქ და იყენებენ Spark-ს.

ამ პოსტში ჩვენ გეტყვით, თუ როგორ გადავჭრით კლასტერის არათანაბარი დატვირთვის პრობლემა ღრუბლის გამოყენებით საკუთარი ავტოსკალერის დაწერით Mail.ru Cloud Solutions.

პრობლემა

ჩვენი კლასტერი არ გამოიყენება ტიპიურ რეჟიმში. განადგურება ძალზე არათანაბარია. მაგალითად, არის პრაქტიკული გაკვეთილები, როდესაც 30-ვე ადამიანი და მასწავლებელი მიდიან კლასტერში და იწყებენ მის გამოყენებას. ან კიდევ, არის ვადამდე დღეები, როდესაც დატვირთვა მნიშვნელოვნად იზრდება. დანარჩენ დროს კლასტერი მუშაობს დატვირთულ რეჟიმში.

Решение №1 – это держать кластер, который будет выдерживать пиковые загрузки, но будет простаивать во все остальное время.

გამოსავალი #2 არის პატარა კლასტერის შენარჩუნება, რომელსაც ხელით ამატებთ კვანძებს კლასების წინ და პიკური დატვირთვის დროს.

გამოსავალი #3 არის პატარა კლასტერის შენახვა და ავტოსკალერის დაწერა, რომელიც მონიტორინგს გაუწევს კლასტერის მიმდინარე დატვირთვას და სხვადასხვა API-ების გამოყენებით დაამატებს და ამოიღებს კვანძებს კლასტერიდან.

ამ პოსტში ვისაუბრებთ გამოსავალზე #3. ეს ავტოსკალერი დიდად არის დამოკიდებული გარე ფაქტორებზე და არა შიდა ფაქტორებზე და პროვაიდერები ხშირად არ აწვდიან მას. ჩვენ ვიყენებთ Mail.ru Cloud Solutions ღრუბლოვან ინფრასტრუქტურას და დავწერეთ ავტოსკალერი MCS API-ის გამოყენებით. და რადგან ჩვენ ვასწავლით მონაცემებთან მუშაობას, გადავწყვიტეთ გვეჩვენებინა, თუ როგორ შეგიძლიათ დაწეროთ მსგავსი ავტოსკალერი თქვენი მიზნებისთვის და გამოიყენოთ იგი თქვენს ღრუბელთან

წინაპირობები

პირველ რიგში, თქვენ უნდა გქონდეთ Hadoop კლასტერი. მაგალითად, ჩვენ ვიყენებთ HDP განაწილებას.

იმისათვის, რომ თქვენი კვანძები სწრაფად დაემატოს და წაიშალოს, თქვენ უნდა გქონდეთ როლების გარკვეული განაწილება კვანძებს შორის.

სამაგისტრო კვანძი. ისე, არაფრის განსაკუთრებული ახსნა არ არის საჭირო: კლასტერის მთავარი კვანძი, რომელზედაც გაშვებულია, მაგალითად, Spark დრაივერი, თუ იყენებთ ინტერაქტიულ რეჟიმს.
თარიღის კვანძი. ეს არის კვანძი, რომელზეც ინახავთ მონაცემებს HDFS-ზე და სადაც ხდება გამოთვლები.
გამოთვლითი კვანძი. ეს არის კვანძი, სადაც თქვენ არ ინახავთ არაფერს HDFS-ზე, მაგრამ სადაც ხდება გამოთვლები.

მნიშვნელოვანი წერტილი. ავტოსკალირება მოხდება მესამე ტიპის კვანძების გამო. თუ დაიწყებთ მეორე ტიპის კვანძების აღებას და დამატებას, რეაგირების სიჩქარე ძალიან დაბალი იქნება - დეკომისაციას და ხელახლა ჩართვისას საათები დასჭირდება თქვენს კლასტერზე. ეს, რა თქმა უნდა, არ არის ის, რასაც ელოდებით ავტომატური სკალირებისგან. ანუ პირველი და მეორე ტიპის კვანძებს არ ვეხებით. ისინი წარმოადგენენ მინიმალურ სიცოცხლისუნარიან კლასტერს, რომელიც იარსებებს პროგრამის განმავლობაში.

ასე რომ, ჩვენი ავტოსკალერი დაწერილია Python 3-ში, იყენებს Ambari API-ს კლასტერული სერვისების სამართავად, იყენებს API Mail.ru Cloud Solutions-დან (MCS) მანქანების გაშვებისა და გაჩერებისთვის.

გადაწყვეტის არქიტექტურა

მოდული autoscaler.py. იგი შეიცავს სამ კლასს: 1) ამბართან მუშაობის ფუნქციები, 2) MCS-თან მუშაობის ფუნქციები, 3) ფუნქციები, რომლებიც უშუალოდ უკავშირდება ავტოსკალერის ლოგიკას.
სკრიპტი observer.py. არსებითად ის შედგება სხვადასხვა წესებისგან: როდის და რა მომენტებში გამოძახება ავტოსკალერის ფუნქციები.
კონფიგურაციის ფაილი 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 – თქვენი კლასტერის სახელი ამბარში,
headers = {'X-Requested-By': 'ambari'}
და შიგნით auth აქ არის თქვენი შესვლა და პაროლი ამბარისთვის: auth = ('login', 'password').

ფუნქცია თავისთავად სხვა არაფერია, თუ არა რამდენიმე ზარი REST API-ით ამბარისკენ. ლოგიკური თვალსაზრისით, ჩვენ ჯერ ვიღებთ კვანძზე გაშვებული სერვისების ჩამონათვალს და შემდეგ ვთხოვთ მოცემულ კლასტერზე, მოცემულ კვანძზე, სერვისების გადატანას სიიდან სახელმწიფოში. INSTALLED. ფუნქციები ყველა სერვისის გაშვებისთვის, კვანძების სახელმწიფოში გადასატანად Maintenance და ა.შ. გამოიყურება მსგავსი - ისინი მხოლოდ რამდენიმე მოთხოვნაა API-ის მეშვეობით.

კლასი მაკს

ასე გამოიყურება კოდის ნაწილი, რომელიც შეიცავს კლასს 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. შემდეგი, ჩვენ უბრალოდ ვაკეთებთ პოსტის მოთხოვნას და გავუშვით ეს მანქანა.

ასე გამოიყურება ჟეტონის მიღების ფუნქცია:

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

ავტოსკალერის კლასი

ეს კლასი შეიცავს ფუნქციებს, რომლებიც დაკავშირებულია თავად ოპერაციულ ლოგიკასთან.

ასე გამოიყურება ამ კლასის კოდის ნაწილი:

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, კვანძების სია, რომლებიც დაშვებულია სკალირებაზე, ასევე კვანძის კონფიგურაციის პარამეტრები: მეხსიერება და CPU გამოყოფილი კვანძისთვის 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 არის კარგი და მოსახერხებელი გადაწყვეტა იმ შემთხვევებისთვის, როდესაც თქვენ განიცდით არათანაბარი კასეტური დატვირთვას. თქვენ ერთდროულად მიაღწევთ სასურველ კასეტურ კონფიგურაციას პიკური დატვირთვისთვის და ამავდროულად არ ინახავთ ამ კლასტერს დატვირთვის დროს, დაზოგავთ ფულს. გარდა ამისა, ეს ყველაფერი ავტომატურად ხდება თქვენი მონაწილეობის გარეშე. თავად ავტოსკალერი სხვა არაფერია, თუ არა კლასტერ მენეჯერის API-სა და ღრუბლოვანი პროვაიდერის API-ს მიმართ მოთხოვნების ნაკრები, რომელიც დაწერილია გარკვეული ლოგიკის მიხედვით. რაც აუცილებლად უნდა გახსოვდეთ, არის კვანძების დაყოფა 3 ტიპად, როგორც ადრე დავწერეთ. და ბედნიერი იქნები.

წყარო: www.habr.com

როგორ გააკეთოთ თქვენი საკუთარი ავტოსკალერი კლასტერისთვის