🥇 Wie man seinen eigenen Autoscaler für ein Cluster erstellt

Hallo! Wir bringen Menschen den Umgang mit Big Data bei. Eine Bildungsmaßnahme zu Big Data ist ohne eigenen Cluster, in dem alle Teilnehmer zusammenarbeiten, kaum vorstellbar. Aus diesem Grund ist er immer Teil unseres Programms 🙂 Wir kümmern uns um die Einrichtung, Feinabstimmung und Administration, während die Teilnehmer dort MapReduce-Jobs ausführen und Spark nutzen.

In diesem Beitrag erzählen wir, wie wir das Problem der ungleichmäßigen Lastverteilung im Cluster gelöst haben, indem wir unseren eigenen Autoscaler im Einsatz hatten, den wir in der Cloud implementiert haben. Mail.ru Cloud Lösungen.

Das Problem

Unser Cluster wird nicht ganz im typischen Modus verwendet. Die Auslastung ist sehr ungleichmäßig. Zum Beispiel gibt es praktische Übungen, bei denen alle 30 Teilnehmer und der Dozent auf den Cluster zugreifen und ihn nutzen. Oder es gibt Tage vor einer Deadline, an denen die Last stark ansteigt. Zu allen anderen Zeiten arbeitet der Cluster unterlastet.

Lösung Nr. 1 – einen Cluster bereitstellen, der hohe Lastspitzen aushält, aber in allen anderen Zeiten ungenutzt bleibt.

Lösung Nr. 2 – einen kleinen Cluster zu betreiben, in den Nodes manuell vor den Aufgaben und während der Spitzenlasten hinzugefügt werden.

Lösung Nr. 3 – einen kleinen Cluster zu betreiben und einen Autoscaler zu schreiben, der die aktuelle Auslastung des Clusters überwacht und selbständig mithilfe verschiedener APIs Nodes zum Cluster hinzufügt oder entfernt.

In diesem Beitrag werden wir über Lösung Nr. 3 sprechen. Ein solcher Autoscaler hängt stark von externen Faktoren ab und nicht von internen, und Anbieter stellen ihn oft nicht zur Verfügung. Wir nutzen die Cloud-Infrastruktur von Mail.ru Cloud Solutions und haben einen Autoscaler geschrieben, der die API von MCS verwendet. Da wir mit Daten arbeiten, möchten wir zeigen, wie Sie einen solchen Autoscaler für Ihre eigenen Zwecke schreiben und mit Ihrer Cloud nutzen können.

Voraussetzungen

Zunächst benötigen Sie ein Hadoop-Cluster. Wir verwenden beispielsweise die Distribution HDP.

Damit Nodes schnell hinzugefügt und entfernt werden können, muss eine bestimmte Rollenverteilung auf den Nodes vorhanden sein.

Master-Node. Hier braucht es keine besondere Erklärung: die Haupt-Node des Clusters, auf der beispielsweise der Spark-Treiber läuft, wenn Sie den interaktiven Modus verwenden.
Data-Node. Dies ist der Knoten, auf dem Ihre Daten im HDFS gespeichert sind und wo die Berechnungen stattfinden.
Compute-Node. Dies ist der Knoten, auf dem keine Daten im HDFS gespeichert sind, aber Berechnungen durchgeführt werden.

Wichtiger Hinweis. Auto-Scaling erfolgt über Knoten des dritten Typs. Wenn Sie anfangen, Knoten des zweiten Typs abzuziehen und hinzuzufügen, wird die Reaktionsgeschwindigkeit stark reduziert – das Dekommissionieren und Rekommissionieren wird Stunden in Ihrem Cluster in Anspruch nehmen. Das ist natürlich nicht das, was man von Auto-Scaling erwartet. Das bedeutet, dass die Knoten des ersten und zweiten Typs unangetastet bleiben. Sie bilden einen minimal funktionsfähigen Cluster, der während der gesamten Programmdauer bestehen bleibt.

Unser Auto-Scaler ist in Python 3 geschrieben und verwendet die Ambari-API zur Verwaltung der Cluster-Services, nutzt die API von Mail.ru Cloud Solutions (MCS) zum Starten und Stoppen von Maschinen.

Architektur der Lösung

Modul autoscaler.py. Darin sind drei Klassen definiert: 1) Funktionen zur Arbeit mit Ambari, 2) Funktionen zur Arbeit mit MCS, 3) Funktionen, die direkt mit der Logik des Auto-Scalers verbunden sind.
Script observer.py. Besteht im Wesentlichen aus verschiedenen Regeln: Wann und zu welchen Zeitpunkten die Funktionen des Auto-Scalers aufgerufen werden.
Die Datei mit den Konfigurationseinstellungen config.py. Sie enthält beispielsweise die Liste der Knoten, die für das Auto-Scaling zugelassen sind, sowie andere Parameter, die beeinflussen, wie lange gewartet werden soll, nachdem ein neuer Knoten hinzugefügt wurde. Dort befinden sich auch die Zeitstempel für den Beginn der Sessions, damit vor der Sitzung die maximal zulässige Klassifizierung des Clusters gestartet wird.

Lassen Sie uns nun einen Blick auf die Codeabschnitte werfen, die sich in den ersten beiden Dateien befinden.

1. Modul autoscaler.py

Klasse Ambari

So sieht der Codeabschnitt aus, der die Klasse enthält 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":"Alle Host-Komponenten stoppen",
                "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 = 'Anfrage akzeptiert'
        else:
            message = req.status_code
        return message

Im obigen Beispiel kann die Implementierung der Funktion betrachtet werden stop_all_services, die alle Dienste auf dem gewünschten Cluster-Knoten stoppt.

Dem Klasseneingang Ambari übergeben Sie:

ambari_url, beispielsweise im Format 'http://localhost:8080/api/v1/clusters/',
cluster_name – der Name Ihres Clusters in Ambari,
headers = {'X-Requested-By': 'ambari'}
und darin auth befindet sich Ihr Benutzername und Passwort für Ambari: auth = ('login', 'password').

Die Funktion besteht lediglich aus ein paar Aufrufen über die REST API zu Ambari. Logisch gesehen erhalten wir zunächst eine Liste von laufenden Diensten auf dem Knoten, und dann bitten wir, diese Dienste im Cluster und auf diesem Knoten in den Zustand INSTALLED. Funktionen zum Starten aller Dienste und zum Wechseln der Knoten in den Status Maintenance und ähnliche sehen ähnlich aus – es sind einfach mehrere Anfragen über die API.

Klasse Mcs

So sieht der Codeabschnitt aus, der die Klasse enthält 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

Dem Klasseneingang Mcs wir übergeben die Projekt-ID in der Cloud und die Benutzer-ID sowie dessen Passwort. In der Funktion vm_turn_on Wir möchten eine der Maschinen einschalten. Die Logik hier ist etwas komplexer. Zu Beginn des Codes rufen wir drei andere Funktionen auf: 1) Wir müssen ein Token erhalten, 2) wir müssen den Hostnamen in den Namen der Maschine im MCS umwandeln, 3) die ID dieser Maschine abrufen. Danach führen wir einfach eine POST-Anfrage durch und starten diese Maschine.

So sieht die Funktion zum Abrufen des Tokens aus:

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

Die Klasse Autoscaler

In dieser Klasse befinden sich Funktionen, die sich auf die Logik der Arbeitsweise beziehen.

So sieht ein Abschnitt des Codes dieser Klasse aus:

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('Decommission-Anfrage angenommen: {0}'.format(flag1))
                    break
            while True:
                time.sleep(5)
                status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
                if status3 == 'INSTALLED':
                    flag3 = True
                    logging.info('Nodemaneger stillgelegt: {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('Wartungsanfrage angenommen: {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('Wartung aktiviert: {0}'.format(flag4))
                    logging.info('Dienste werden gestoppt')
                    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 ist ausgeschaltet: {0}'.format(flag5))
                    break
            if flag1 and flag2 and flag3 and flag4 and flag5:
                message = 'Erfolg'
                logging.info('Scale-Down abgeschlossen')
                logging.info('Abkühlzeit hat begonnen. Warten Sie einige Minuten')
        return message

Wir akzeptieren Klassen als Eingabe Ambari und Mcs, eine Liste von Knoten, die für das Skalieren erlaubt sind, sowie die Konfigurationsparameter der Knoten: Speicher und CPU, die dem Knoten in YARN zugewiesen sind. Es gibt auch zwei interne Parameter q_ram, q_cpu, die Warteschlangen darstellen. Mit ihnen speichern wir die Werte der aktuellen Clusterlast. Wenn wir sehen, dass in den letzten 5 Minuten eine konstante hohe Last bestand, entscheiden wir, dass ein weiterer Knoten dem Cluster hinzugefügt werden muss. Dasselbe gilt für den Zustand einer Unterauslastung des Clusters.

Im obigen Code ist ein Beispiel für eine Funktion aufgeführt, die eine Maschine aus dem Cluster entfernt und sie in der Cloud stoppt. Zunächst erfolgt das Decommissioning YARN Nodemanager, dann wird der Modus aktiviert Maintenance, dann stoppen wir alle Dienste auf der Maschine und schalten die virtuelle Maschine in der Cloud aus.

2. Das Skript observer.py

Beispielcode daraus:

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} wurde erfolgreich hochskaliert".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)

Darin prüfen wir, ob die Bedingungen für die Skalierung des Clusters erfüllt sind und ob überhaupt Maschinen in der Reserve vorhanden sind. Wir erhalten den Hostnamen einer von ihnen, fügen ihn zum Cluster hinzu und veröffentlichen eine Nachricht darüber in Slack unseres Teams. Danach startet cooldown_period, in der wir nichts hinzufügen oder entfernen, sondern nur die Auslastung überwachen. Wenn sie sich stabilisiert hat und innerhalb des optimalen Wertebereichs liegt, setzen wir die Überwachung einfach fort. Wenn jedoch eine Node nicht ausreicht, fügen wir eine weitere hinzu.

In Fällen, in denen wir bereits wissen, dass eine Node nicht ausreicht, starten wir sofort alle verfügbaren Nodes und halten sie bis zum Ende der Sitzung aktiv. Dies geschieht durch eine Liste mit Zeitstempeln der Sitzungen.

Fazit

Der Autoscaler ist eine ausgezeichnete und benutzerfreundliche Lösung für Fälle, in denen eine ungleichmäßige Auslastung des Clusters vorliegt. So erreichen Sie gleichzeitig die benötigte Clusterkonfiguration für Spitzenlasten, ohne das Cluster während Unterauslastung aufrechterhalten zu müssen, wodurch Sie Kosten sparen. Außerdem geschieht alles automatisiert, ohne Ihr Eingreifen. Der Autoscaler selbst ist nichts anderes als eine Reihe von API-Anfragen an den Cluster-Manager und den Cloud-Anbieter, die nach bestimmten Logiken formuliert sind. Wichtig dabei ist das Angebot von Knoten in 3 Typen, wie wir zuvor erwähnt haben. Dann steht Ihrem Erfolg nichts mehr im Weg.

Quelle: habr.com