334 lines
12 KiB
Python
Executable File
334 lines
12 KiB
Python
Executable File
#!/usr/bin/env python
|
|
#
|
|
# Copyright 2014 Huawei Technologies Co. Ltd
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""deploy cluster from csv file."""
|
|
import ast
|
|
import copy
|
|
import csv
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
from multiprocessing import Process
|
|
from multiprocessing import Queue
|
|
from optparse import OptionParser
|
|
|
|
try:
|
|
from compass.apiclient.restful import Client
|
|
except ImportError:
|
|
curr_dir = os.path.dirname(os.path.realpath(__file__))
|
|
apiclient_dir = os.path.dirname(curr_dir) + '/compass/apiclient'
|
|
sys.path.append(apiclient_dir)
|
|
from restful import Client
|
|
|
|
|
|
DELIMITER = ","
|
|
|
|
# Sqlite tables
|
|
TABLES = {
|
|
'switch_config': {'columns': ['id', 'ip', 'filter_port']},
|
|
'switch': {'columns': ['id', 'ip', 'credential_data']},
|
|
'machine': {'columns': ['id', 'mac', 'port', 'vlan', 'switch_id']},
|
|
'cluster': {'columns': ['id', 'name', 'security_config',
|
|
'networking_config', 'partition_config',
|
|
'adapter_id', 'state']},
|
|
'cluster_host': {'columns': ['id', 'cluster_id', 'hostname', 'machine_id',
|
|
'config_data', 'state']},
|
|
'adapter': {'columns': ['id', 'name', 'os', 'target_system']},
|
|
'role': {'columns': ['id', 'name', 'target_system', 'description']}
|
|
}
|
|
|
|
|
|
def start(csv_dir, compass_url):
|
|
"""Start deploy both failed clusters and new clusters."""
|
|
# Get clusters and hosts data from CSV
|
|
clusters_data = get_csv('cluster.csv', csv_dir)
|
|
hosts_data = get_csv('cluster_host.csv', csv_dir)
|
|
data = {}
|
|
for cluster in clusters_data:
|
|
tmp = {}
|
|
tmp['cluster_data'] = cluster
|
|
tmp['hosts_data'] = []
|
|
data[cluster['id']] = tmp
|
|
|
|
for host in hosts_data:
|
|
cluster_id = host['cluster_id']
|
|
if cluster_id not in data:
|
|
print ("Unknown cluster_id=%s of the host! host_id=%s!"
|
|
% (cluster_id, host['id']))
|
|
sys.exit(1)
|
|
|
|
data[cluster_id]['hosts_data'].append(host)
|
|
|
|
apiClient = _APIClient(compass_url)
|
|
results_q = Queue()
|
|
ps = []
|
|
for elem in data:
|
|
cluster_data = data[elem]['cluster_data']
|
|
hosts_data = data[elem]['hosts_data']
|
|
p = Process(target=apiClient.execute,
|
|
args=(cluster_data, hosts_data, results_q))
|
|
ps.append(p)
|
|
p.start()
|
|
|
|
for p in ps:
|
|
p.join()
|
|
|
|
progress_file = '/'.join((csv_dir, 'progress.csv'))
|
|
write_progress_to_file(results_q, progress_file)
|
|
|
|
|
|
def write_progress_to_file(results_q, progress_file):
|
|
cluster_headers = ['cluster_id', 'progress_url']
|
|
host_headers = ['host_id', 'progress_url']
|
|
|
|
with open(progress_file, 'wb') as f:
|
|
print "Writing all progress information to %s......" % progress_file
|
|
writer = csv.writer(f, delimiter=DELIMITER, quoting=csv.QUOTE_MINIMAL)
|
|
while not results_q.empty():
|
|
record = results_q.get()
|
|
hosts = []
|
|
cluster = [record['deployment']['cluster']['cluster_id'],
|
|
record['deployment']['cluster']['url']]
|
|
writer.writerow(cluster_headers)
|
|
writer.writerow(cluster)
|
|
|
|
for elem in record['deployment']['hosts']:
|
|
host = [elem['host_id'], elem['url']]
|
|
hosts.append(host)
|
|
|
|
writer.writerow(host_headers)
|
|
writer.writerows(hosts)
|
|
print "Done!\n"
|
|
|
|
|
|
def get_csv(fname, csv_dir):
|
|
"""Parse csv files into python variables.
|
|
|
|
.. note::
|
|
all nested fields in db will be assembled.
|
|
|
|
:param fname: CSV file name
|
|
:param csv_dir: CSV files directory
|
|
|
|
:returns: list of dict which key is column name and value is its data.
|
|
"""
|
|
headers = []
|
|
rows = []
|
|
file_dir = '/'.join((csv_dir, fname))
|
|
with open(file_dir) as f:
|
|
reader = csv.reader(f, delimiter=DELIMITER, quoting=csv.QUOTE_MINIMAL)
|
|
headers = reader.next()
|
|
rows = [x for x in reader]
|
|
|
|
result = []
|
|
for row in rows:
|
|
data = {}
|
|
for col_name, value in zip(headers, row):
|
|
if re.match(r'^[\d]+$', value):
|
|
# the value should be an integer
|
|
value = int(value)
|
|
elif re.match(r'^\[(\'\w*\'){1}(\s*,\s*\'\w*\')*\]$', value):
|
|
# the value should be a list
|
|
value = ast.literal_eval(value)
|
|
elif value == 'None':
|
|
value = ''
|
|
|
|
if col_name.find('.') > 0:
|
|
tmp_result = {}
|
|
tmp_result[col_name.split('.')[-1]] = value
|
|
keys = col_name.split('.')[::-1][1:]
|
|
for key in keys:
|
|
tmp = {}
|
|
tmp[key] = tmp_result
|
|
tmp_result = tmp
|
|
merge_dict(data, tmp_result)
|
|
else:
|
|
data[col_name] = value
|
|
|
|
result.append(data)
|
|
|
|
return result
|
|
|
|
|
|
def merge_dict(lhs, rhs, override=True):
|
|
"""Merge nested right dict into left nested dict recursively.
|
|
|
|
:param lhs: dict to be merged into.
|
|
:type lhs: dict
|
|
:param rhs: dict to merge from.
|
|
:type rhs: dict
|
|
:param override: the value in rhs overide the value in left if True.
|
|
:type override: str
|
|
|
|
:raises: TypeError if lhs or rhs is not a dict.
|
|
"""
|
|
if not rhs:
|
|
return
|
|
|
|
if not isinstance(lhs, dict):
|
|
raise TypeError('lhs type is %s while expected is dict' % type(lhs),
|
|
lhs)
|
|
|
|
if not isinstance(rhs, dict):
|
|
raise TypeError('rhs type is %s while expected is dict' % type(rhs),
|
|
rhs)
|
|
|
|
for key, value in rhs.items():
|
|
if isinstance(value, dict) and key in lhs and isinstance(lhs[key],
|
|
dict):
|
|
merge_dict(lhs[key], value, override)
|
|
else:
|
|
if override or key not in lhs:
|
|
lhs[key] = copy.deepcopy(value)
|
|
|
|
|
|
class _APIClient(Client):
|
|
def __init__(self, url, headers=None, proxies=None, stream=None):
|
|
super(_APIClient, self).__init__(url, headers, proxies, stream)
|
|
|
|
def set_cluster_resource(self, cluster_id, resource, data):
|
|
url = "/clusters/%d/%s" % (cluster_id, resource)
|
|
return self._put(url, data=data)
|
|
|
|
def execute(self, cluster_data, hosts_data, resp_results):
|
|
"""The process includes creating or updating a cluster.
|
|
|
|
The cluster configuration, add or update a host in the cluster,
|
|
and deploy the updated hosts.
|
|
|
|
:param cluster_data: the dictionary of cluster data
|
|
"""
|
|
cluster_id = cluster_data['id']
|
|
code, resp = self.get_cluster(cluster_id)
|
|
if code == 404:
|
|
# Create a new cluster
|
|
name = cluster_data['name']
|
|
adapter_id = cluster_data['adapter_id']
|
|
code, resp = self.add_cluster(name, adapter_id)
|
|
|
|
if code != 200:
|
|
print ("Failed to create the cluster which name is "
|
|
"%s!\nError message: %s" % (name, resp['message']))
|
|
sys.exit(1)
|
|
|
|
# Update the config(security, networking, partition) of the cluster
|
|
security_req = {}
|
|
networking_req = {}
|
|
partition_req = {}
|
|
|
|
security_req['security'] = cluster_data['security_config']
|
|
networking_req['networking'] = cluster_data['networking_config']
|
|
partition_req['partition'] = cluster_data['partition_config']
|
|
|
|
print "Update Security config......."
|
|
code, resp = self.set_cluster_resource(cluster_id, 'security',
|
|
security_req)
|
|
if code != 200:
|
|
print ("Failed to update Security config for cluster id=%s!\n"
|
|
"Error message: " % (cluster_id, resp['message']))
|
|
sys.exit(1)
|
|
|
|
print "Update Networking config......."
|
|
code, resp = self.set_cluster_resource(cluster_id, 'networking',
|
|
networking_req)
|
|
if code != 200:
|
|
print ("Failed to update Networking config for cluster id=%s!\n"
|
|
"Error message: %s" % (cluster_id, resp['message']))
|
|
sys.exit(1)
|
|
|
|
print "Update Partition config......."
|
|
code, resp = self.set_cluster_resource(cluster_id, 'partition',
|
|
partition_req)
|
|
if code != 200:
|
|
print ("Failed to update Partition config for cluster id=%s!\n"
|
|
"Error message: " % (cluster_id, resp['message']))
|
|
sys.exit(1)
|
|
|
|
deploy_list = []
|
|
deploy_hosts_data = []
|
|
|
|
machines_list = []
|
|
new_hosts_data = []
|
|
for record in hosts_data:
|
|
if record['state'] and int(record['deploy_action']):
|
|
deploy_list.append(record['id'])
|
|
deploy_hosts_data.append(record)
|
|
|
|
elif int(record['deploy_action']):
|
|
machines_list.append(record['machine_id'])
|
|
new_hosts_data.append(record)
|
|
|
|
if machines_list:
|
|
# add new hosts to the cluster
|
|
code, resp = self.add_hosts(cluster_id, machines_list)
|
|
if code != 200:
|
|
print ("Failed to add hosts to the cluster id=%s!\n"
|
|
"Error message: %s.\nfailed hosts are %s"
|
|
% (cluster_id, resp['message'], resp['failedMachines']))
|
|
sys.exit(1)
|
|
|
|
for record, host in zip(new_hosts_data, resp['cluster_hosts']):
|
|
record['id'] = host['id']
|
|
deploy_list.append(host['id'])
|
|
deploy_hosts_data.append(record)
|
|
|
|
# Update the config of each host in the cluster
|
|
for host in deploy_hosts_data:
|
|
req = {}
|
|
host_id = host['id']
|
|
print "Updating the config of host id=%s" % host['id']
|
|
req['hostname'] = host['hostname']
|
|
req.update(host['config_data'])
|
|
code, resp = self.update_host_config(int(host_id), raw_data=req)
|
|
if code != 200:
|
|
print ("Failed to update the config of the host id=%s!\n"
|
|
"Error message: %s" % (host_id, resp['message']))
|
|
sys.exit(1)
|
|
|
|
# Start to deploy the cluster
|
|
print "Start to deploy the cluster!....."
|
|
deploy_req = {"deploy": deploy_list}
|
|
code, resp = self.deploy_hosts(cluster_id, raw_data=deploy_req)
|
|
print "---Cluster Info---"
|
|
print "cluster_id url"
|
|
print (" %s %s"
|
|
% (resp['deployment']['cluster']['cluster_id'],
|
|
resp['deployment']['cluster']['url']))
|
|
print "---Hosts Info-----"
|
|
print "host_id url"
|
|
for host in resp['deployment']['hosts']:
|
|
print " %s %s" % (host['host_id'], host['url'])
|
|
print "---------------------------------------------------------------"
|
|
print "\n"
|
|
resp_results.put(resp)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
usage = "usage: %prog [options]"
|
|
parser = OptionParser(usage)
|
|
|
|
parser.add_option("-d", "--csv-dir", dest="csv_dir",
|
|
help="The directory of CSV files used for depolyment")
|
|
parser.add_option("-u", "--compass-url", dest="compass_url",
|
|
help="The URL of Compass server")
|
|
(options, args) = parser.parse_args()
|
|
|
|
if not os.exists(options.csv_dir):
|
|
print "Cannot find the directory: %s" % options.csv_dir
|
|
|
|
start(options.csv_dir, options.compass_url)
|