របៀបបង្កើត autoscaler ផ្ទាល់ខ្លួនរបស់អ្នកសម្រាប់ចង្កោម

សួស្តី! យើងបណ្តុះបណ្តាលមនុស្សឱ្យធ្វើការជាមួយទិន្នន័យធំ។ វាមិនអាចទៅរួចទេក្នុងការស្រមៃមើលកម្មវិធីអប់រំលើទិន្នន័យធំដោយគ្មានចង្កោមរបស់វា ដែលអ្នកចូលរួមទាំងអស់ធ្វើការជាមួយគ្នា។ សម្រាប់ហេតុផលនេះ កម្មវិធីរបស់យើងតែងតែមានវា :) យើងបានចូលរួមក្នុងការកំណត់រចនាសម្ព័ន្ធ ការលៃតម្រូវ និងការគ្រប់គ្រងរបស់វា ហើយបុរសចាប់ផ្តើមការងារ MapReduce ដោយផ្ទាល់នៅទីនោះ ហើយប្រើ Spark ។

នៅក្នុងការប្រកាសនេះ យើងនឹងប្រាប់អ្នកពីរបៀបដែលយើងដោះស្រាយបញ្ហានៃការផ្ទុកចង្កោមមិនស្មើគ្នាដោយការសរសេរ autoscaler របស់យើងផ្ទាល់ដោយប្រើ cloud ដំណោះស្រាយពពក Mail.ru.

បញ្ហា

ចង្កោមរបស់យើងមិនត្រូវបានប្រើក្នុងទម្រង់ធម្មតាទេ។ ការបោះចោលគឺមិនស្មើគ្នាខ្លាំង។ ជាឧទាហរណ៍ មានថ្នាក់អនុវត្ត នៅពេលដែលមនុស្សទាំង 30 នាក់ និងគ្រូម្នាក់ទៅចង្កោម ហើយចាប់ផ្តើមប្រើវា។ ឬម្តងទៀត មានថ្ងៃមុនថ្ងៃផុតកំណត់ នៅពេលដែលបន្ទុកកើនឡើងយ៉ាងខ្លាំង។ នៅសល់នៃពេលវេលាដែលចង្កោមដំណើរការក្នុងរបៀបមិនផ្ទុក។

ដំណោះស្រាយទី 1 គឺរក្សាចង្កោមដែលនឹងទប់ទល់នឹងបន្ទុកខ្ពស់បំផុត ប៉ុន្តែនឹងនៅទំនេរអស់ពេលដែលនៅសល់។

ដំណោះស្រាយ #2 គឺត្រូវរក្សាចង្កោមតូចមួយ ដែលអ្នកបន្ថែមថ្នាំងដោយដៃមុនថ្នាក់ និងកំឡុងពេលផ្ទុកខ្ពស់បំផុត។

ដំណោះស្រាយ #3 គឺរក្សាចង្កោមតូចមួយ ហើយសរសេរ autoscaler ដែលនឹងត្រួតពិនិត្យបន្ទុកបច្ចុប្បន្ននៃ cluster ហើយដោយប្រើ APIs ផ្សេងៗ បន្ថែម និងលុបថ្នាំងចេញពី cluster ។

នៅក្នុងការប្រកាសនេះ យើងនឹងនិយាយអំពីដំណោះស្រាយលេខ 3 ។ ឧបករណ៍វាស់ស្ទង់ស្វ័យប្រវត្តិនេះពឹងផ្អែកខ្លាំងលើកត្តាខាងក្រៅជាជាងកត្តាខាងក្នុង ហើយអ្នកផ្តល់សេវាជារឿយៗមិនផ្តល់វាទេ។ យើងប្រើប្រាស់ហេដ្ឋារចនាសម្ព័ន្ធពពក Mail.ru Cloud Solutions ហើយសរសេរកម្មវិធីធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិដោយប្រើ MCS API ។ ហើយចាប់តាំងពីយើងបង្រៀនពីរបៀបធ្វើការជាមួយទិន្នន័យ យើងបានសម្រេចចិត្តបង្ហាញពីរបៀបដែលអ្នកអាចសរសេរ autoscaler ស្រដៀងគ្នាសម្រាប់គោលបំណងផ្ទាល់ខ្លួនរបស់អ្នក ហើយប្រើវាជាមួយ cloud របស់អ្នក

តម្រូវការជាមុន

ដំបូងអ្នកត្រូវតែមានចង្កោម Hadoop ។ ឧទាហរណ៍ យើងប្រើការចែកចាយ HDP ។

ដើម្បីឱ្យថ្នាំងរបស់អ្នកត្រូវបានបន្ថែម និងដកចេញយ៉ាងឆាប់រហ័ស អ្នកត្រូវតែមានការចែកចាយតួនាទីជាក់លាក់ក្នុងចំណោមថ្នាំង។

1. ថ្នាំងមេ។ ជាការប្រសើរណាស់ គ្មានអ្វីចាំបាច់ដើម្បីពន្យល់នៅទីនេះទេ៖ ថ្នាំងសំខាន់នៃចង្កោម ដែលឧទាហរណ៍ កម្មវិធីបញ្ជា Spark ត្រូវបានបើកដំណើរការ ប្រសិនបើអ្នកប្រើរបៀបអន្តរកម្ម។
2. ថ្នាំងកាលបរិច្ឆេទ។ នេះគឺជាថ្នាំងដែលអ្នករក្សាទុកទិន្នន័យនៅលើ HDFS និងកន្លែងដែលការគណនាកើតឡើង។
3. ថ្នាំងកុំព្យូទ័រ។ នេះគឺជាថ្នាំងដែលអ្នកមិនរក្សាទុកអ្វីនៅលើ HDFS ប៉ុន្តែជាកន្លែងដែលការគណនាកើតឡើង។

ចំណុចសំខាន់។ ការធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិនឹងកើតឡើងដោយសារថ្នាំងនៃប្រភេទទីបី។ ប្រសិនបើអ្នកចាប់ផ្តើមទទួលយក និងបន្ថែមថ្នាំងនៃប្រភេទទីពីរ ល្បឿនឆ្លើយតបនឹងទាបណាស់ - ការបញ្ឈប់ និងបញ្ចូលឡើងវិញនឹងចំណាយពេលច្រើនម៉ោងនៅលើចង្កោមរបស់អ្នក។ ជាការពិតណាស់ នេះមិនមែនជាអ្វីដែលអ្នករំពឹងពីការធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិនោះទេ។ នោះគឺយើងមិនប៉ះថ្នាំងនៃប្រភេទទីមួយនិងទីពីរទេ។ ពួកវានឹងតំណាងឱ្យក្រុមដែលអាចសម្រេចបានអប្បបរមាដែលនឹងមានពេញមួយរយៈពេលនៃកម្មវិធី។

ដូច្នេះ autoscaler របស់យើងត្រូវបានសរសេរនៅក្នុង Python 3 ប្រើប្រាស់ Ambari API ដើម្បីគ្រប់គ្រងសេវាកម្ម cluster ប្រើប្រាស់ API ពី Mail.ru Cloud Solutions (MCS) សម្រាប់ចាប់ផ្តើម និងបញ្ឈប់ម៉ាស៊ីន។

ស្ថាបត្យកម្មដំណោះស្រាយ

1. ម៉ូឌុល autoscaler.py. វាមានបីថ្នាក់៖ 1) មុខងារសម្រាប់ធ្វើការជាមួយ Ambari, 2) មុខងារសម្រាប់ធ្វើការជាមួយ MCS, 3) មុខងារដែលទាក់ទងដោយផ្ទាល់ទៅនឹងតក្កវិជ្ជារបស់ autoscaler ។
2. ស្គ្រីប observer.py. សំខាន់វាមានច្បាប់ផ្សេងៗគ្នា៖ ពេលណា និងពេលណាមួយដើម្បីហៅមុខងារ autoscaler ។
3. ឯកសារកំណត់រចនាសម្ព័ន្ធ 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":"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) យើងត្រូវបំប្លែង hostname ទៅជាឈ្មោះរបស់ម៉ាស៊ីនក្នុង 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បញ្ជីថ្នាំងដែលត្រូវបានអនុញ្ញាតសម្រាប់ការធ្វើមាត្រដ្ឋាន ក៏ដូចជាប៉ារ៉ាម៉ែត្រកំណត់រចនាសម្ព័ន្ធថ្នាំង៖ អង្គចងចាំ និងស៊ីភីយូដែលបានបែងចែកទៅថ្នាំងក្នុង 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} 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