Hello! We train people to work with big data. It's hard to imagine a big data educational program without its own cluster, where all participants work together. For this reason, our program always includes one 🙂 We handle its setup, tuning, and administration, while the participants run MapReduce jobs and utilize Spark.
In this post, we will explain how we solved the issue of uneven cluster load by creating our own auto-scaler using the cloud. .
The Problem
Our cluster is used in a somewhat atypical mode. The utilization is highly uneven. For example, there are practical sessions when all 30 students and the instructor access the cluster and start using it. Again, there are days before deadlines when the load significantly increases. During all other times, the cluster operates in an underloaded state.
Solution #1 is to maintain a cluster that can handle peak loads but will be idle at all other times.
Solution #2 is to keep a small cluster and manually add nodes before classes and during peak loads.
Solution #3 is to maintain a small cluster and write an auto-scaler that monitors the current load of the cluster and automatically adds and removes nodes using various APIs.
In this post, we will discuss solution #3. Such an auto-scaler heavily depends on external factors rather than internal ones, and providers often do not offer them. We use the cloud infrastructure from Mail.ru Cloud Solutions and have written an auto-scaler using the MCS API. Since we teach data management, we decided to demonstrate how you can create a similar auto-scaler for your needs and use it with your own cloud.
Prerequisites
First, you need to have a Hadoop cluster. For instance, we use the HDP distribution.
For nodes to be able to quickly add and remove, you must have a specific distribution of roles across the nodes.
- Master node. There's not much to explain here: it's the main node of the cluster where, for example, the Spark driver runs if you're using interactive mode.
- Data node. This is the node where your data is stored on HDFS and where computations take place.
- Compute node. This is a node where nothing is stored on HDFS, but computations take place.
Important point. Autoscaling will occur based on type three nodes. If you start removing and adding type two nodes, the response speed will be significantly low – decommissioning and recommissioning will take hours on your cluster. This, of course, is not what you expect from autoscaling. So we do not touch type one and two nodes. They will represent a minimally viable cluster that will exist throughout the program's duration.
So, our autoscaler is written in Python 3, uses the Ambari API to manage cluster services, and utilizes (MCS) to start and stop machines.
Solution Architecture
- Module
autoscaler.py. It contains three classes: 1) functions for working with Ambari, 2) functions for working with MCS, 3) functions directly related to the autoscaler's logic. - Script
observer.py. It essentially consists of different rules: when and at what moments to call the autoscaler's functions. - A file with configuration parameters
config.py. It contains, for example, a list of nodes allowed for autoscaling and other parameters, such as how long to wait from the moment a new node was added. It also includes timestamps for the start of sessions, so the maximum allowed cluster configuration is launched before the session begins.
Now let's look at code snippets from the first two files.
1. autoscaler.py module
Ambari class
Here is a snippet of code containing the Ambari class:
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 messageThe example above shows an implementation of the function stop_all_services, which stops all services on the specified node of the cluster.
To the class Ambari class you pass:
ambari_url, for example, in the format'http://localhost:8080/api/v1/clusters/',cluster_name– the name of your cluster in Ambari,headers = {'X-Requested-By': 'ambari'}- and inside
authare your login and password for Ambari:auth = ('login', 'password').
The function itself consists of just a couple of calls to the Ambari REST API. Logically, we first obtain a list of the running services on the node, and then request to set the services from this list to the state INSTALLED. Functions for starting all services, transitioning nodes to the state Maintenance and others look similar – they are just several API requests.
Class Mcs
Here is a snippet of code containing the 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_codeTo the class Mcs we pass the project ID within the cloud and the user ID, as well as their password. In the function vm_turn_on We want to start one of the machines. The logic here is a bit more complex. At the beginning of the code, there are calls to three other functions: 1) we need to obtain a token, 2) we need to convert the hostname to the machine name in MCS, 3) get the ID of this machine. Then we simply make a post request and start this machine.
Here is the function for obtaining the token:
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.tokenAutoscaler Class
This class contains functions related to the core logic of its operation.
Here is a snippet of this class's code:
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 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 messageWe accept classes as input Ambari class and Mcs, a list of nodes allowed for scaling, as well as configuration parameters for the nodes: memory and CPU allocated per node in YARN. There are also 2 internal parameters q_ram, q_cpu, which are queues. We use them to store the current load values of the cluster. If we see that there has been a consistently high load for the last 5 minutes, we decide that we need to add +1 node to the cluster. The same applies to the state of underload in the cluster.
The code above shows an example of a function that removes a machine from the cluster and stops it in the cloud. Initially, decommissioning occurs YARN Nodemanager, then maintenance mode is enabled Maintenance, then we stop all services on the machine and turn off the virtual machine in the cloud.
2. The script observer.py
Example code from there:
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)Here we check if the conditions for scaling up the cluster's capacity are met and if there are any machines available in reserve; we get the hostname of one of them, add it to the cluster, and publish a message about this in our team's Slack. After that, we start the cooldown_period, during which we do not add or remove anything from the cluster, but simply monitor the load. If it stabilizes and stays within the optimal load range, we just continue monitoring. If one node is not enough, we add another.
In cases where we have a session coming up, we already know for sure that one node will not be enough, so we immediately start all available nodes and keep them active until the end of the session. This is done using a list of timestamps for the sessions.
Conclusion
An autoscaler is a good and convenient solution for situations where you have uneven cluster load. You simultaneously achieve the required cluster configuration for peak loads without keeping this cluster running during low load periods, saving costs. Plus, all this happens automatically without your involvement. The autoscaler is essentially just a set of requests to the cluster manager's API and the cloud provider's API, written according to a specific logic. One thing to definitely remember is the division of nodes into 3 types, as we mentioned earlier. And happiness will come to you.
Source: habr.com
