Sahara aims to provide users with simple means to provision a Hadoop cluster by specifying several parameters like Hadoop version, cluster topology, nodes hardware details and a few more.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

deploy.py 25KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709
  1. # Copyright (c) 2015 Mirantis Inc.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12. # implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  15. import functools
  16. import telnetlib # nosec
  17. from oslo_log import log as logging
  18. from oslo_utils import uuidutils
  19. from sahara import conductor
  20. from sahara import context
  21. from sahara.i18n import _
  22. from sahara.plugins.ambari import client as ambari_client
  23. from sahara.plugins.ambari import common as p_common
  24. from sahara.plugins.ambari import configs
  25. from sahara.plugins.ambari import ha_helper
  26. from sahara.plugins import kerberos
  27. from sahara.plugins import utils as plugin_utils
  28. from sahara.topology import topology_helper as t_helper
  29. from sahara.utils import cluster_progress_ops as cpo
  30. from sahara.utils import poll_utils
  31. LOG = logging.getLogger(__name__)
  32. conductor = conductor.API
  33. repo_id_map = {
  34. "2.3": {
  35. "HDP": "HDP-2.3",
  36. "HDP-UTILS": "HDP-UTILS-1.1.0.20"
  37. },
  38. "2.4": {
  39. "HDP": "HDP-2.4",
  40. "HDP-UTILS": "HDP-UTILS-1.1.0.20"
  41. },
  42. "2.5": {
  43. "HDP": "HDP-2.5",
  44. "HDP-UTILS": "HDP-UTILS-1.1.0.21"
  45. },
  46. "2.6": {
  47. "HDP": "HDP-2.6",
  48. "HDP-UTILS": "HDP-UTILS-1.1.0.22"
  49. },
  50. }
  51. os_type_map = {
  52. "centos6": "redhat6",
  53. "redhat6": "redhat6",
  54. "centos7": "redhat7",
  55. "redhat7": "redhat7",
  56. "ubuntu14": "ubuntu14"
  57. }
  58. @cpo.event_wrapper(True, step=_("Set up Ambari management console"),
  59. param=('cluster', 0))
  60. def setup_ambari(cluster):
  61. LOG.debug("Set up Ambari management console")
  62. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  63. with ambari.remote() as r:
  64. sudo = functools.partial(r.execute_command, run_as_root=True)
  65. sudo("rngd -r /dev/urandom -W 4096")
  66. sudo("ambari-server setup -s -j"
  67. " `cut -f2 -d \"=\" /etc/profile.d/99-java.sh`", timeout=1800)
  68. redirect_file = "/tmp/%s" % uuidutils.generate_uuid()
  69. sudo("service ambari-server start >{rfile} && "
  70. "cat {rfile} && rm {rfile}".format(rfile=redirect_file))
  71. LOG.debug("Ambari management console installed")
  72. def setup_agents(cluster, instances=None):
  73. LOG.debug("Set up Ambari agents")
  74. manager_address = plugin_utils.get_instance(
  75. cluster, p_common.AMBARI_SERVER).fqdn()
  76. if not instances:
  77. instances = plugin_utils.get_instances(cluster)
  78. _setup_agents(instances, manager_address)
  79. def _setup_agents(instances, manager_address):
  80. cpo.add_provisioning_step(
  81. instances[0].cluster.id, _("Set up Ambari agents"), len(instances))
  82. with context.ThreadGroup() as tg:
  83. for inst in instances:
  84. tg.spawn("hwx-agent-setup-%s" % inst.id,
  85. _setup_agent, inst, manager_address)
  86. LOG.debug("Ambari agents have been installed")
  87. def _disable_repos_on_inst(instance):
  88. with context.set_current_instance_id(instance_id=instance.instance_id):
  89. with instance.remote() as r:
  90. sudo = functools.partial(r.execute_command, run_as_root=True)
  91. if r.get_os_distrib() == "ubuntu":
  92. sudo("mv /etc/apt/sources.list /etc/apt/sources.list.tmp")
  93. else:
  94. tmp_name = "/tmp/yum.repos.d-%s" % instance.instance_id[:8]
  95. # moving to other folder
  96. sudo("mv /etc/yum.repos.d/ {fold_name}".format(
  97. fold_name=tmp_name))
  98. sudo("mkdir /etc/yum.repos.d")
  99. def disable_repos(cluster):
  100. if configs.use_base_repos_needed(cluster):
  101. LOG.debug("Using base repos")
  102. return
  103. instances = plugin_utils.get_instances(cluster)
  104. with context.ThreadGroup() as tg:
  105. for inst in instances:
  106. tg.spawn("disable-repos-%s" % inst.instance_name,
  107. _disable_repos_on_inst, inst)
  108. @cpo.event_wrapper(True)
  109. def _setup_agent(instance, ambari_address):
  110. with instance.remote() as r:
  111. sudo = functools.partial(r.execute_command, run_as_root=True)
  112. r.replace_remote_string("/etc/ambari-agent/conf/ambari-agent.ini",
  113. "localhost", ambari_address)
  114. try:
  115. sudo("ambari-agent start")
  116. except Exception as e:
  117. # workaround for ubuntu, because on ubuntu the ambari agent
  118. # starts automatically after image boot
  119. msg = _("Restart of ambari-agent is needed for host {}, "
  120. "reason: {}").format(instance.fqdn(), e)
  121. LOG.exception(msg)
  122. sudo("ambari-agent restart")
  123. # for correct installing packages
  124. r.update_repository()
  125. @cpo.event_wrapper(True, step=_("Wait Ambari accessible"),
  126. param=('cluster', 0))
  127. def wait_ambari_accessible(cluster):
  128. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  129. kwargs = {"host": ambari.management_ip, "port": 8080}
  130. poll_utils.poll(_check_port_accessible, kwargs=kwargs, timeout=300)
  131. def _check_port_accessible(host, port):
  132. try:
  133. conn = telnetlib.Telnet(host, port)
  134. conn.close()
  135. return True
  136. except IOError:
  137. return False
  138. def resolve_package_conflicts(cluster, instances=None):
  139. if not instances:
  140. instances = plugin_utils.get_instances(cluster)
  141. for instance in instances:
  142. with instance.remote() as r:
  143. if r.get_os_distrib() == 'ubuntu':
  144. try:
  145. r.execute_command(
  146. "apt-get remove -y libmysql-java", run_as_root=True)
  147. except Exception:
  148. LOG.warning("Can't remove libmysql-java, "
  149. "it's probably not installed")
  150. def _prepare_ranger(cluster):
  151. ranger = plugin_utils.get_instance(cluster, p_common.RANGER_ADMIN)
  152. if not ranger:
  153. return
  154. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  155. with ambari.remote() as r:
  156. sudo = functools.partial(r.execute_command, run_as_root=True)
  157. sudo("ambari-server setup --jdbc-db=mysql "
  158. "--jdbc-driver=/usr/share/java/mysql-connector-java.jar")
  159. init_db_template = (
  160. "create user 'root'@'%' identified by '{password}';\n"
  161. "set password for 'root'@'localhost' = password('{password}');")
  162. password = uuidutils.generate_uuid()
  163. extra = cluster.extra.to_dict() if cluster.extra else {}
  164. extra["ranger_db_password"] = password
  165. ctx = context.ctx()
  166. conductor.cluster_update(ctx, cluster, {"extra": extra})
  167. with ranger.remote() as r:
  168. sudo = functools.partial(r.execute_command, run_as_root=True)
  169. # TODO(sreshetnyak): add ubuntu support
  170. sudo("yum install -y mysql-server")
  171. sudo("service mysqld start")
  172. r.write_file_to("/tmp/init.sql",
  173. init_db_template.format(password=password))
  174. sudo("mysql < /tmp/init.sql")
  175. sudo("rm /tmp/init.sql")
  176. @cpo.event_wrapper(True, step=_("Prepare Hive"), param=('cluster', 0))
  177. def prepare_hive(cluster):
  178. hive = plugin_utils.get_instance(cluster, p_common.HIVE_SERVER)
  179. if not hive:
  180. return
  181. with hive.remote() as r:
  182. r.execute_command(
  183. 'sudo su - -c "hadoop fs -mkdir /user/oozie/conf" hdfs')
  184. r.execute_command(
  185. 'sudo su - -c "hadoop fs -copyFromLocal '
  186. '/etc/hive/conf/hive-site.xml '
  187. '/user/oozie/conf/hive-site.xml" hdfs')
  188. @cpo.event_wrapper(True, step=_("Update default Ambari password"),
  189. param=('cluster', 0))
  190. def update_default_ambari_password(cluster):
  191. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  192. new_password = uuidutils.generate_uuid()
  193. with ambari_client.AmbariClient(ambari) as client:
  194. client.update_user_password("admin", "admin", new_password)
  195. extra = cluster.extra.to_dict() if cluster.extra else {}
  196. extra["ambari_password"] = new_password
  197. ctx = context.ctx()
  198. conductor.cluster_update(ctx, cluster, {"extra": extra})
  199. cluster = conductor.cluster_get(ctx, cluster.id)
  200. @cpo.event_wrapper(True, step=_("Wait registration of hosts"),
  201. param=('cluster', 0))
  202. def wait_host_registration(cluster, instances):
  203. with _get_ambari_client(cluster) as client:
  204. kwargs = {"client": client, "instances": instances}
  205. poll_utils.poll(_check_host_registration, kwargs=kwargs, timeout=600)
  206. def _check_host_registration(client, instances):
  207. hosts = client.get_registered_hosts()
  208. registered_host_names = [h["Hosts"]["host_name"] for h in hosts]
  209. for instance in instances:
  210. if instance.fqdn() not in registered_host_names:
  211. return False
  212. return True
  213. @cpo.event_wrapper(True, step=_("Set up HDP repositories"),
  214. param=('cluster', 0))
  215. def _set_up_hdp_repos(cluster, hdp_repo, hdp_utils_repo):
  216. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  217. pv = cluster.hadoop_version
  218. repos = repo_id_map[pv]
  219. with _get_ambari_client(cluster) as client:
  220. os_type = os_type_map[client.get_host_info(ambari.fqdn())["os_type"]]
  221. if hdp_repo:
  222. client.set_up_mirror(pv, os_type, repos["HDP"], hdp_repo)
  223. if hdp_utils_repo:
  224. client.set_up_mirror(pv, os_type, repos["HDP-UTILS"],
  225. hdp_utils_repo)
  226. def set_up_hdp_repos(cluster):
  227. hdp_repo = configs.get_hdp_repo_url(cluster)
  228. hdp_utils_repo = configs.get_hdp_utils_repo_url(cluster)
  229. if hdp_repo or hdp_utils_repo:
  230. _set_up_hdp_repos(cluster, hdp_repo, hdp_utils_repo)
  231. def get_kdc_server(cluster):
  232. return plugin_utils.get_instance(
  233. cluster, p_common.AMBARI_SERVER)
  234. def _prepare_kerberos(cluster, instances=None):
  235. if instances is None:
  236. kerberos.deploy_infrastructure(cluster, get_kdc_server(cluster))
  237. kerberos.prepare_policy_files(cluster)
  238. else:
  239. server = None
  240. if not kerberos.using_existing_kdc(cluster):
  241. server = get_kdc_server(cluster)
  242. kerberos.setup_clients(cluster, server)
  243. kerberos.prepare_policy_files(cluster)
  244. def prepare_kerberos(cluster, instances=None):
  245. if kerberos.is_kerberos_security_enabled(cluster):
  246. _prepare_kerberos(cluster, instances)
  247. def _serialize_mit_kdc_kerberos_env(cluster):
  248. return {
  249. 'kerberos-env': {
  250. "realm": kerberos.get_realm_name(cluster),
  251. "kdc_type": "mit-kdc",
  252. "kdc_host": kerberos.get_kdc_host(
  253. cluster, get_kdc_server(cluster)),
  254. "admin_server_host": kerberos.get_kdc_host(
  255. cluster, get_kdc_server(cluster)),
  256. 'encryption_types': 'aes256-cts-hmac-sha1-96',
  257. 'ldap_url': '', 'container_dn': '',
  258. }
  259. }
  260. def _serialize_krb5_configs(cluster):
  261. return {
  262. "krb5-conf": {
  263. "properties_attributes": {},
  264. "properties": {
  265. "manage_krb5_conf": "false"
  266. }
  267. }
  268. }
  269. def _get_credentials(cluster):
  270. return [{
  271. "alias": "kdc.admin.credential",
  272. "principal": kerberos.get_admin_principal(cluster),
  273. "key": kerberos.get_server_password(cluster),
  274. "type": "TEMPORARY"
  275. }]
  276. def get_host_group_components(cluster, processes):
  277. result = []
  278. for proc in processes:
  279. result.append({'name': proc})
  280. return result
  281. @cpo.event_wrapper(True, step=_("Create Ambari blueprint"),
  282. param=('cluster', 0))
  283. def create_blueprint(cluster):
  284. _prepare_ranger(cluster)
  285. cluster = conductor.cluster_get(context.ctx(), cluster.id)
  286. host_groups = []
  287. for ng in cluster.node_groups:
  288. procs = p_common.get_ambari_proc_list(ng)
  289. procs.extend(p_common.get_clients(cluster))
  290. for instance in ng.instances:
  291. hg = {
  292. "name": instance.instance_name,
  293. "configurations": configs.get_instance_params(instance),
  294. "components": get_host_group_components(cluster, procs)
  295. }
  296. host_groups.append(hg)
  297. bp = {
  298. "Blueprints": {
  299. "stack_name": "HDP",
  300. "stack_version": cluster.hadoop_version,
  301. },
  302. "host_groups": host_groups,
  303. "configurations": configs.get_cluster_params(cluster)
  304. }
  305. if kerberos.is_kerberos_security_enabled(cluster):
  306. bp['configurations'].extend([
  307. _serialize_mit_kdc_kerberos_env(cluster),
  308. _serialize_krb5_configs(cluster)
  309. ])
  310. bp['Blueprints']['security'] = {'type': 'KERBEROS'}
  311. general_configs = cluster.cluster_configs.get("general", {})
  312. if (general_configs.get(p_common.NAMENODE_HA) or
  313. general_configs.get(p_common.RESOURCEMANAGER_HA) or
  314. general_configs.get(p_common.HBASE_REGIONSERVER_HA)):
  315. bp = ha_helper.update_bp_ha_common(cluster, bp)
  316. if general_configs.get(p_common.NAMENODE_HA):
  317. bp = ha_helper.update_bp_for_namenode_ha(cluster, bp)
  318. if general_configs.get(p_common.RESOURCEMANAGER_HA):
  319. bp = ha_helper.update_bp_for_resourcemanager_ha(cluster, bp)
  320. if general_configs.get(p_common.HBASE_REGIONSERVER_HA):
  321. bp = ha_helper.update_bp_for_hbase_ha(cluster, bp)
  322. with _get_ambari_client(cluster) as client:
  323. return client.create_blueprint(cluster.name, bp)
  324. def _build_ambari_cluster_template(cluster):
  325. cl_tmpl = {
  326. "blueprint": cluster.name,
  327. "default_password": uuidutils.generate_uuid(),
  328. "host_groups": []
  329. }
  330. if cluster.use_autoconfig:
  331. strategy = configs.get_auto_configuration_strategy(cluster)
  332. cl_tmpl["config_recommendation_strategy"] = strategy
  333. if kerberos.is_kerberos_security_enabled(cluster):
  334. cl_tmpl["credentials"] = _get_credentials(cluster)
  335. cl_tmpl["security"] = {"type": "KERBEROS"}
  336. topology = _get_topology_data(cluster)
  337. for ng in cluster.node_groups:
  338. for instance in ng.instances:
  339. host = {"fqdn": instance.fqdn()}
  340. if t_helper.is_data_locality_enabled():
  341. host["rack_info"] = topology[instance.instance_name]
  342. cl_tmpl["host_groups"].append({
  343. "name": instance.instance_name,
  344. "hosts": [host]
  345. })
  346. return cl_tmpl
  347. @cpo.event_wrapper(True, step=_("Start cluster"), param=('cluster', 0))
  348. def start_cluster(cluster):
  349. ambari_template = _build_ambari_cluster_template(cluster)
  350. with _get_ambari_client(cluster) as client:
  351. req_id = client.create_cluster(cluster.name, ambari_template)["id"]
  352. client.wait_ambari_request(req_id, cluster.name)
  353. @cpo.event_wrapper(True)
  354. def _add_host_to_cluster(instance, client):
  355. client.add_host_to_cluster(instance)
  356. def add_new_hosts(cluster, instances):
  357. with _get_ambari_client(cluster) as client:
  358. cpo.add_provisioning_step(
  359. cluster.id, _("Add new hosts"), len(instances))
  360. for inst in instances:
  361. _add_host_to_cluster(inst, client)
  362. @cpo.event_wrapper(True, step=_("Generate config groups"),
  363. param=('cluster', 0))
  364. def manage_config_groups(cluster, instances):
  365. groups = []
  366. for instance in instances:
  367. groups.extend(configs.get_config_group(instance))
  368. with _get_ambari_client(cluster) as client:
  369. client.create_config_group(cluster, groups)
  370. @cpo.event_wrapper(True, step=_("Cleanup config groups"),
  371. param=('cluster', 0))
  372. def cleanup_config_groups(cluster, instances):
  373. to_remove = set()
  374. for instance in instances:
  375. cfg_name = "%s:%s" % (cluster.name, instance.instance_name)
  376. to_remove.add(cfg_name)
  377. with _get_ambari_client(cluster) as client:
  378. config_groups = client.get_config_groups(cluster)
  379. for group in config_groups['items']:
  380. cfg_id = group['ConfigGroup']['id']
  381. detailed = client.get_detailed_config_group(cluster, cfg_id)
  382. cfg_name = detailed['ConfigGroup']['group_name']
  383. # we have config group per host
  384. if cfg_name in to_remove:
  385. client.remove_config_group(cluster, cfg_id)
  386. @cpo.event_wrapper(True, step=_("Regenerate keytabs for Kerberos"),
  387. param=('cluster', 0))
  388. def _regenerate_keytabs(cluster):
  389. with _get_ambari_client(cluster) as client:
  390. alias = "kdc.admin.credential"
  391. try:
  392. client.get_credential(cluster.name, alias)
  393. except ambari_client.AmbariNotFound:
  394. # credentials are missing
  395. data = {
  396. 'Credential': {
  397. "principal": kerberos.get_admin_principal(cluster),
  398. "key": kerberos.get_server_password(cluster),
  399. "type": "TEMPORARY"
  400. }
  401. }
  402. client.import_credential(cluster.name, alias, data)
  403. req_id = client.regenerate_keytabs(cluster.name)
  404. client.wait_ambari_request(req_id, cluster.name)
  405. @cpo.event_wrapper(True, step=_("Install services on hosts"),
  406. param=('cluster', 0))
  407. def _install_services_to_hosts(cluster, instances):
  408. requests_ids = []
  409. with _get_ambari_client(cluster) as client:
  410. clients = p_common.get_clients(cluster)
  411. for instance in instances:
  412. services = p_common.get_ambari_proc_list(instance.node_group)
  413. services.extend(clients)
  414. for service in services:
  415. client.add_service_to_host(instance, service)
  416. requests_ids.append(
  417. client.start_service_on_host(
  418. instance, service, 'INSTALLED'))
  419. client.wait_ambari_requests(requests_ids, cluster.name)
  420. @cpo.event_wrapper(True, step=_("Start services on hosts"),
  421. param=('cluster', 0))
  422. def _start_services_on_hosts(cluster, instances):
  423. with _get_ambari_client(cluster) as client:
  424. # all services added and installed, let's start them
  425. requests_ids = []
  426. for instance in instances:
  427. services = p_common.get_ambari_proc_list(instance.node_group)
  428. services.extend(p_common.ALL_LIST)
  429. for service in services:
  430. requests_ids.append(
  431. client.start_service_on_host(
  432. instance, service, 'STARTED'))
  433. client.wait_ambari_requests(requests_ids, cluster.name)
  434. def manage_host_components(cluster, instances):
  435. _install_services_to_hosts(cluster, instances)
  436. if kerberos.is_kerberos_security_enabled(cluster):
  437. _regenerate_keytabs(cluster)
  438. _start_services_on_hosts(cluster, instances)
  439. @cpo.event_wrapper(True, step=_("Decommission NodeManagers and DataNodes"),
  440. param=('cluster', 0))
  441. def decommission_hosts(cluster, instances):
  442. nodemanager_instances = filter(
  443. lambda i: p_common.NODEMANAGER in i.node_group.node_processes,
  444. instances)
  445. if len(nodemanager_instances) > 0:
  446. decommission_nodemanagers(cluster, nodemanager_instances)
  447. datanode_instances = filter(
  448. lambda i: p_common.DATANODE in i.node_group.node_processes,
  449. instances)
  450. if len(datanode_instances) > 0:
  451. decommission_datanodes(cluster, datanode_instances)
  452. def decommission_nodemanagers(cluster, instances):
  453. with _get_ambari_client(cluster) as client:
  454. client.decommission_nodemanagers(cluster.name, instances)
  455. def decommission_datanodes(cluster, instances):
  456. with _get_ambari_client(cluster) as client:
  457. client.decommission_datanodes(cluster.name, instances)
  458. def restart_namenode(cluster, instance):
  459. with _get_ambari_client(cluster) as client:
  460. client.restart_namenode(cluster.name, instance)
  461. def restart_resourcemanager(cluster, instance):
  462. with _get_ambari_client(cluster) as client:
  463. client.restart_resourcemanager(cluster.name, instance)
  464. @cpo.event_wrapper(True, step=_("Restart NameNodes and ResourceManagers"),
  465. param=('cluster', 0))
  466. def restart_nns_and_rms(cluster):
  467. nns = plugin_utils.get_instances(cluster, p_common.NAMENODE)
  468. for nn in nns:
  469. restart_namenode(cluster, nn)
  470. rms = plugin_utils.get_instances(cluster, p_common.RESOURCEMANAGER)
  471. for rm in rms:
  472. restart_resourcemanager(cluster, rm)
  473. def restart_service(cluster, service_name):
  474. with _get_ambari_client(cluster) as client:
  475. client.restart_service(cluster.name, service_name)
  476. @cpo.event_wrapper(True, step=_("Remove hosts"), param=('cluster', 0))
  477. def remove_services_from_hosts(cluster, instances):
  478. for inst in instances:
  479. LOG.debug("Stopping and removing processes from host %s", inst.fqdn())
  480. _remove_services_from_host(cluster, inst)
  481. LOG.debug("Removing the host %s", inst.fqdn())
  482. _remove_host(cluster, inst)
  483. def _remove_services_from_host(cluster, instance):
  484. with _get_ambari_client(cluster) as client:
  485. hdp_processes = client.list_host_processes(cluster.name, instance)
  486. for proc in hdp_processes:
  487. LOG.debug("Stopping process %(proc)s on host %(fqdn)s ",
  488. {'proc': proc, 'fqdn': instance.fqdn()})
  489. client.stop_process_on_host(cluster.name, instance, proc)
  490. LOG.debug("Removing process %(proc)s from host %(fqdn)s ",
  491. {'proc': proc, 'fqdn': instance.fqdn()})
  492. client.remove_process_from_host(cluster.name, instance, proc)
  493. _wait_all_processes_removed(cluster, instance)
  494. def _remove_host(cluster, inst):
  495. with _get_ambari_client(cluster) as client:
  496. client.delete_host(cluster.name, inst)
  497. def _wait_all_processes_removed(cluster, instance):
  498. with _get_ambari_client(cluster) as client:
  499. while True:
  500. hdp_processes = client.list_host_processes(cluster.name, instance)
  501. if not hdp_processes:
  502. return
  503. context.sleep(5)
  504. def _get_ambari_client(cluster):
  505. ambari = plugin_utils.get_instance(cluster, p_common.AMBARI_SERVER)
  506. password = cluster.extra["ambari_password"]
  507. return ambari_client.AmbariClient(ambari, password=password)
  508. def _get_topology_data(cluster):
  509. if not t_helper.is_data_locality_enabled():
  510. return {}
  511. LOG.warning("Node group awareness is not implemented in YARN yet "
  512. "so enable_hypervisor_awareness set to False "
  513. "explicitly")
  514. return t_helper.generate_topology_map(cluster, is_node_awareness=False)
  515. @cpo.event_wrapper(True)
  516. def _configure_topology_data(cluster, inst, client):
  517. topology = _get_topology_data(cluster)
  518. client.set_rack_info_for_instance(
  519. cluster.name, inst, topology[inst.instance_name])
  520. @cpo.event_wrapper(True, step=_("Restart HDFS and MAPREDUCE2 services"),
  521. param=('cluster', 0))
  522. def _restart_hdfs_and_mapred_services(cluster, client):
  523. client.restart_service(cluster.name, p_common.HDFS_SERVICE)
  524. client.restart_service(cluster.name, p_common.MAPREDUCE2_SERVICE)
  525. def configure_rack_awareness(cluster, instances):
  526. if not t_helper.is_data_locality_enabled():
  527. return
  528. with _get_ambari_client(cluster) as client:
  529. cpo.add_provisioning_step(
  530. cluster.id, _("Configure rack awareness"), len(instances))
  531. for inst in instances:
  532. _configure_topology_data(cluster, inst, client)
  533. _restart_hdfs_and_mapred_services(cluster, client)
  534. @cpo.event_wrapper(True)
  535. def _add_hadoop_swift_jar(instance, new_jar):
  536. with instance.remote() as r:
  537. code, out = r.execute_command(
  538. "test -f %s" % new_jar, raise_when_error=False)
  539. if code == 0:
  540. # get ambari hadoop version (e.g.: 2.7.1.2.3.4.0-3485)
  541. code, amb_hadoop_version = r.execute_command(
  542. "sudo hadoop version | grep 'Hadoop' | awk '{print $2}'")
  543. amb_hadoop_version = amb_hadoop_version.strip()
  544. # get special code of ambari hadoop version(e.g.:2.3.4.0-3485)
  545. amb_code = '.'.join(amb_hadoop_version.split('.')[3:])
  546. origin_jar = (
  547. "/usr/hdp/{}/hadoop-mapreduce/hadoop-openstack-{}.jar".format(
  548. amb_code, amb_hadoop_version))
  549. r.execute_command("sudo cp {} {}".format(new_jar, origin_jar))
  550. else:
  551. LOG.warning("The {jar_file} file cannot be found "
  552. "in the {dir} directory so Keystone API v3 "
  553. "is not enabled for this cluster."
  554. .format(jar_file="hadoop-openstack.jar",
  555. dir="/opt"))
  556. def add_hadoop_swift_jar(instances):
  557. new_jar = "/opt/hadoop-openstack.jar"
  558. cpo.add_provisioning_step(instances[0].cluster.id,
  559. _("Add Hadoop Swift jar to instances"),
  560. len(instances))
  561. for inst in instances:
  562. _add_hadoop_swift_jar(inst, new_jar)
  563. def deploy_kerberos_principals(cluster, instances=None):
  564. if not kerberos.is_kerberos_security_enabled(cluster):
  565. return
  566. if instances is None:
  567. instances = plugin_utils.get_instances(cluster)
  568. mapper = {
  569. 'hdfs': plugin_utils.instances_with_services(
  570. instances, [p_common.SECONDARY_NAMENODE, p_common.NAMENODE,
  571. p_common.DATANODE, p_common.JOURNAL_NODE]),
  572. 'spark': plugin_utils.instances_with_services(
  573. instances, [p_common.SPARK_JOBHISTORYSERVER]),
  574. 'oozie': plugin_utils.instances_with_services(
  575. instances, [p_common.OOZIE_SERVER]),
  576. }
  577. kerberos.create_keytabs_for_map(cluster, mapper)