CustomServiceOrchestrator.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666
  1. #!/usr/bin/env python
  2. '''
  3. Licensed to the Apache Software Foundation (ASF) under one
  4. or more contributor license agreements. See the NOTICE file
  5. distributed with this work for additional information
  6. regarding copyright ownership. The ASF licenses this file
  7. to you under the Apache License, Version 2.0 (the
  8. "License"); you may not use this file except in compliance
  9. with the License. You may obtain a copy of the License at
  10. http://www.apache.org/licenses/LICENSE-2.0
  11. Unless required by applicable law or agreed to in writing, software
  12. distributed under the License is distributed on an "AS IS" BASIS,
  13. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. See the License for the specific language governing permissions and
  15. limitations under the License.
  16. '''
  17. import logging
  18. import os
  19. import ambari_simplejson as json
  20. import sys
  21. from ambari_commons import shell
  22. import threading
  23. from FileCache import FileCache
  24. from AgentException import AgentException
  25. from PythonExecutor import PythonExecutor
  26. from PythonReflectiveExecutor import PythonReflectiveExecutor
  27. from resource_management.libraries.functions.log_process_information import log_process_information
  28. from resource_management.core.utils import PasswordString
  29. import subprocess
  30. import Constants
  31. import hostname
  32. logger = logging.getLogger()
  33. class CustomServiceOrchestrator():
  34. """
  35. Executes a command for custom service. stdout and stderr are written to
  36. tmpoutfile and to tmperrfile respectively.
  37. """
  38. SCRIPT_TYPE_PYTHON = "PYTHON"
  39. COMMAND_TYPE = "commandType"
  40. COMMAND_NAME_STATUS = "STATUS"
  41. COMMAND_NAME_SECURITY_STATUS = "SECURITY_STATUS"
  42. CUSTOM_ACTION_COMMAND = 'ACTIONEXECUTE'
  43. CUSTOM_COMMAND_COMMAND = 'CUSTOM_COMMAND'
  44. PRE_HOOK_PREFIX="before"
  45. POST_HOOK_PREFIX="after"
  46. HOSTS_LIST_KEY = "all_hosts"
  47. PING_PORTS_KEY = "all_ping_ports"
  48. RACKS_KEY = "all_racks"
  49. IPV4_ADDRESSES_KEY = "all_ipv4_ips"
  50. AMBARI_SERVER_HOST = "ambari_server_host"
  51. AMBARI_SERVER_PORT = "ambari_server_port"
  52. AMBARI_SERVER_USE_SSL = "ambari_server_use_ssl"
  53. FREQUENT_COMMANDS = [COMMAND_NAME_SECURITY_STATUS, COMMAND_NAME_STATUS]
  54. DONT_DEBUG_FAILURES_FOR_COMMANDS = FREQUENT_COMMANDS
  55. REFLECTIVELY_RUN_COMMANDS = FREQUENT_COMMANDS # -- commands which run a lot and often (this increases their speed)
  56. DONT_BACKUP_LOGS_FOR_COMMANDS = FREQUENT_COMMANDS
  57. # Path where hadoop credential JARS will be available
  58. DEFAULT_CREDENTIAL_SHELL_LIB_PATH = '/var/lib/ambari-agent/cred/lib'
  59. DEFAULT_CREDENTIAL_CONF_DIR = '/var/lib/ambari-agent/cred/conf'
  60. DEFAULT_CREDENTIAL_SHELL_CMD = 'org.apache.hadoop.security.alias.CredentialShell'
  61. # The property name used by the hadoop credential provider
  62. CREDENTIAL_PROVIDER_PROPERTY_NAME = 'hadoop.security.credential.provider.path'
  63. # Property name for credential store class path
  64. CREDENTIAL_STORE_CLASS_PATH_NAME = 'credentialStoreClassPath'
  65. def __init__(self, config, controller):
  66. self.config = config
  67. self.tmp_dir = config.get('agent', 'prefix')
  68. self.force_https_protocol = config.get_force_https_protocol()
  69. self.exec_tmp_dir = Constants.AGENT_TMP_DIR
  70. self.file_cache = FileCache(config)
  71. self.status_commands_stdout = os.path.join(self.tmp_dir,
  72. 'status_command_stdout.txt')
  73. self.status_commands_stderr = os.path.join(self.tmp_dir,
  74. 'status_command_stderr.txt')
  75. self.public_fqdn = hostname.public_hostname(config)
  76. # cache reset will be called on every agent registration
  77. controller.registration_listeners.append(self.file_cache.reset)
  78. # Construct the hadoop credential lib JARs path
  79. self.credential_shell_lib_path = os.path.join(config.get('security', 'credential_lib_dir',
  80. self.DEFAULT_CREDENTIAL_SHELL_LIB_PATH), '*')
  81. self.credential_conf_dir = config.get('security', 'credential_conf_dir', self.DEFAULT_CREDENTIAL_CONF_DIR)
  82. self.credential_shell_cmd = config.get('security', 'credential_shell_cmd', self.DEFAULT_CREDENTIAL_SHELL_CMD)
  83. # Clean up old status command files if any
  84. try:
  85. os.unlink(self.status_commands_stdout)
  86. os.unlink(self.status_commands_stderr)
  87. except OSError:
  88. pass # Ignore fail
  89. self.commands_in_progress_lock = threading.RLock()
  90. self.commands_in_progress = {}
  91. def map_task_to_process(self, task_id, processId):
  92. with self.commands_in_progress_lock:
  93. logger.debug('Maps taskId=%s to pid=%s' % (task_id, processId))
  94. self.commands_in_progress[task_id] = processId
  95. def cancel_command(self, task_id, reason):
  96. with self.commands_in_progress_lock:
  97. if task_id in self.commands_in_progress.keys():
  98. pid = self.commands_in_progress.get(task_id)
  99. self.commands_in_progress[task_id] = reason
  100. logger.info("Canceling command with taskId = {tid}, " \
  101. "reason - {reason} . Killing process {pid}"
  102. .format(tid=str(task_id), reason=reason, pid=pid))
  103. log_process_information(logger)
  104. shell.kill_process_with_children(pid)
  105. else:
  106. logger.warn("Unable to find process associated with taskId = %s" % task_id)
  107. def get_py_executor(self, forced_command_name):
  108. """
  109. Wrapper for unit testing
  110. :return:
  111. """
  112. if forced_command_name in self.REFLECTIVELY_RUN_COMMANDS:
  113. return PythonReflectiveExecutor(self.tmp_dir, self.config)
  114. else:
  115. return PythonExecutor(self.tmp_dir, self.config)
  116. def getProviderDirectory(self, service_name):
  117. """
  118. Gets the path to the service conf folder where the JCEKS file will be created.
  119. :param service_name: Name of the service, for example, HIVE
  120. :return: lower case path to the service conf folder
  121. """
  122. # The stack definition scripts of the service can move the
  123. # JCEKS file around to where it wants, which is usually
  124. # /etc/<service_name>/conf
  125. conf_dir = os.path.join(self.credential_conf_dir, service_name.lower())
  126. return conf_dir
  127. def getConfigTypeCredentials(self, commandJson):
  128. """
  129. Gets the affected config types for the service in this command
  130. with the password aliases and values.
  131. Input:
  132. {
  133. "config-type1" : {
  134. "password_key_name1":"password_value_name1",
  135. "password_key_name2":"password_value_name2",
  136. :
  137. },
  138. "config-type2" : {
  139. "password_key_name1":"password_value_name1",
  140. "password_key_name2":"password_value_name2",
  141. :
  142. },
  143. :
  144. }
  145. Output:
  146. {
  147. "config-type1" : {
  148. "alias1":"password1",
  149. "alias2":"password2",
  150. :
  151. },
  152. "config-type2" : {
  153. "alias1":"password1",
  154. "alias2":"password2",
  155. :
  156. },
  157. :
  158. }
  159. If password_key_name is the same as password_value_name, then password_key_name is the password alias itself.
  160. The value it points to is the password value.
  161. If password_key_name is not the same as the password_value_name, then password_key_name points to the alias.
  162. The value is pointed to by password_value_name.
  163. For example:
  164. Input:
  165. {
  166. "oozie-site" : {"oozie.service.JPAService.jdbc.password" : "oozie.service.JPAService.jdbc.password"},
  167. "admin-properties" {"db_user":"db_password", "ranger.jpa.jdbc.credential.alias:ranger-admin-site" : "db_password"}
  168. }
  169. Output:
  170. {
  171. "oozie-site" : {"oozie.service.JPAService.jdbc.password" : "MyOozieJdbcPassword"},
  172. "admin-properties" {"rangerdba" : "MyRangerDbaPassword", "rangeradmin":"MyRangerDbaPassword"},
  173. }
  174. :param commandJson:
  175. :return:
  176. """
  177. configtype_credentials = {}
  178. if 'configuration_credentials' in commandJson:
  179. for config_type, password_properties in commandJson['configuration_credentials'].items():
  180. if config_type in commandJson['configurations']:
  181. value_names = []
  182. config = commandJson['configurations'][config_type]
  183. credentials = {}
  184. for key_name, value_name in password_properties.items():
  185. if key_name == value_name:
  186. if value_name in config:
  187. # password name is the alias
  188. credentials[key_name] = config[value_name]
  189. value_names.append(value_name) # Gather the value_name for deletion
  190. else:
  191. keyname_keyconfig = key_name.split(':')
  192. key_name = keyname_keyconfig[0]
  193. # if the key is in another configuration (cross reference),
  194. # get the value of the key from that configuration
  195. if (len(keyname_keyconfig) > 1):
  196. if keyname_keyconfig[1] not in commandJson['configurations']:
  197. continue
  198. key_config = commandJson['configurations'][keyname_keyconfig[1]]
  199. else:
  200. key_config = config
  201. if key_name in key_config and value_name in config:
  202. # password name points to the alias
  203. credentials[key_config[key_name]] = config[value_name]
  204. value_names.append(value_name) # Gather the value_name for deletion
  205. if len(credentials) > 0:
  206. configtype_credentials[config_type] = credentials
  207. logger.info("Identifying config {0} for CS: ".format(config_type))
  208. for value_name in value_names:
  209. # Remove the clear text password
  210. config.pop(value_name, None)
  211. return configtype_credentials
  212. def generateJceks(self, commandJson):
  213. """
  214. Generates the JCEKS file with passwords for the service specified in commandJson
  215. :param commandJson: command JSON
  216. :return: An exit value from the external process that generated the JCEKS file. None if
  217. there are no passwords in the JSON.
  218. """
  219. cmd_result = None
  220. roleCommand = None
  221. if 'roleCommand' in commandJson:
  222. roleCommand = commandJson['roleCommand']
  223. task_id = None
  224. if 'taskId' in commandJson:
  225. task_id = commandJson['taskId']
  226. logger.info('Generating the JCEKS file: roleCommand={0} and taskId = {1}'.format(roleCommand, task_id))
  227. # Set up the variables for the external command to generate a JCEKS file
  228. java_home = commandJson['hostLevelParams']['java_home']
  229. java_bin = '{java_home}/bin/java'.format(java_home=java_home)
  230. cs_lib_path = self.credential_shell_lib_path
  231. serviceName = commandJson['serviceName']
  232. # Gather the password values and remove them from the configuration
  233. configtype_credentials = self.getConfigTypeCredentials(commandJson)
  234. # CS is enabled but no config property is available for this command
  235. if len(configtype_credentials) == 0:
  236. logger.info("Credential store is enabled but no property are found that can be encrypted.")
  237. commandJson['credentialStoreEnabled'] = "false"
  238. for config_type, credentials in configtype_credentials.items():
  239. config = commandJson['configurations'][config_type]
  240. file_path = os.path.join(self.getProviderDirectory(serviceName), "{0}.jceks".format(config_type))
  241. if os.path.exists(file_path):
  242. os.remove(file_path)
  243. provider_path = 'jceks://file{file_path}'.format(file_path=file_path)
  244. logger.info('provider_path={0}'.format(provider_path))
  245. for alias, pwd in credentials.items():
  246. logger.debug("config={0}".format(config))
  247. protected_pwd = PasswordString(pwd)
  248. # Generate the JCEKS file
  249. cmd = (java_bin, '-cp', cs_lib_path, self.credential_shell_cmd, 'create',
  250. alias, '-value', protected_pwd, '-provider', provider_path)
  251. logger.info(cmd)
  252. cmd_result = subprocess.call(cmd)
  253. logger.info('cmd_result = {0}'.format(cmd_result))
  254. os.chmod(file_path, 0644) # group and others should have read access so that the service user can read
  255. # Add JCEKS provider path instead
  256. config[self.CREDENTIAL_PROVIDER_PROPERTY_NAME] = provider_path
  257. config[self.CREDENTIAL_STORE_CLASS_PATH_NAME] = cs_lib_path
  258. return cmd_result
  259. def runCommand(self, command, tmpoutfile, tmperrfile, forced_command_name=None,
  260. override_output_files=True, retry=False):
  261. """
  262. forced_command_name may be specified manually. In this case, value, defined at
  263. command json, is ignored.
  264. """
  265. try:
  266. script_type = command['commandParams']['script_type']
  267. script = command['commandParams']['script']
  268. timeout = int(command['commandParams']['command_timeout'])
  269. if 'hostLevelParams' in command and 'jdk_location' in command['hostLevelParams']:
  270. server_url_prefix = command['hostLevelParams']['jdk_location']
  271. else:
  272. server_url_prefix = command['commandParams']['jdk_location']
  273. task_id = "status"
  274. try:
  275. task_id = command['taskId']
  276. command_name = command['roleCommand']
  277. except KeyError:
  278. pass # Status commands have no taskId
  279. if forced_command_name is not None: # If not supplied as an argument
  280. command_name = forced_command_name
  281. if command_name == self.CUSTOM_ACTION_COMMAND:
  282. base_dir = self.file_cache.get_custom_actions_base_dir(server_url_prefix)
  283. script_tuple = (os.path.join(base_dir, 'scripts', script), base_dir)
  284. hook_dir = None
  285. else:
  286. if command_name == self.CUSTOM_COMMAND_COMMAND:
  287. command_name = command['hostLevelParams']['custom_command']
  288. # forces a hash challenge on the directories to keep them updated, even
  289. # if the return type is not used
  290. self.file_cache.get_host_scripts_base_dir(server_url_prefix)
  291. hook_dir = self.file_cache.get_hook_base_dir(command, server_url_prefix)
  292. base_dir = self.file_cache.get_service_base_dir(command, server_url_prefix)
  293. self.file_cache.get_custom_resources_subdir(command, server_url_prefix)
  294. script_path = self.resolve_script_path(base_dir, script)
  295. script_tuple = (script_path, base_dir)
  296. tmpstrucoutfile = os.path.join(self.tmp_dir,
  297. "structured-out-{0}.json".format(task_id))
  298. # We don't support anything else yet
  299. if script_type.upper() != self.SCRIPT_TYPE_PYTHON:
  300. message = "Unknown script type {0}".format(script_type)
  301. raise AgentException(message)
  302. # Execute command using proper interpreter
  303. handle = None
  304. if command.has_key('__handle'):
  305. handle = command['__handle']
  306. handle.on_background_command_started = self.map_task_to_process
  307. del command['__handle']
  308. # If command contains credentialStoreEnabled, then
  309. # generate the JCEKS file for the configurations.
  310. credentialStoreEnabled = False
  311. if 'credentialStoreEnabled' in command:
  312. credentialStoreEnabled = (command['credentialStoreEnabled'] == "true")
  313. if credentialStoreEnabled == True:
  314. self.generateJceks(command)
  315. json_path = self.dump_command_to_json(command, retry)
  316. pre_hook_tuple = self.resolve_hook_script_path(hook_dir,
  317. self.PRE_HOOK_PREFIX, command_name, script_type)
  318. post_hook_tuple = self.resolve_hook_script_path(hook_dir,
  319. self.POST_HOOK_PREFIX, command_name, script_type)
  320. py_file_list = [pre_hook_tuple, script_tuple, post_hook_tuple]
  321. # filter None values
  322. filtered_py_file_list = [i for i in py_file_list if i]
  323. logger_level = logging.getLevelName(logger.level)
  324. # Executing hooks and script
  325. ret = None
  326. from ActionQueue import ActionQueue
  327. if command.has_key('commandType') and command['commandType'] == ActionQueue.BACKGROUND_EXECUTION_COMMAND and len(filtered_py_file_list) > 1:
  328. raise AgentException("Background commands are supported without hooks only")
  329. python_executor = self.get_py_executor(forced_command_name)
  330. backup_log_files = not command_name in self.DONT_BACKUP_LOGS_FOR_COMMANDS
  331. log_out_files = self.config.get("logging","log_out_files", default="0") != "0"
  332. for py_file, current_base_dir in filtered_py_file_list:
  333. log_info_on_failure = not command_name in self.DONT_DEBUG_FAILURES_FOR_COMMANDS
  334. script_params = [command_name, json_path, current_base_dir, tmpstrucoutfile, logger_level, self.exec_tmp_dir,
  335. self.force_https_protocol]
  336. if log_out_files:
  337. script_params.append("-o")
  338. ret = python_executor.run_file(py_file, script_params,
  339. tmpoutfile, tmperrfile, timeout,
  340. tmpstrucoutfile, self.map_task_to_process,
  341. task_id, override_output_files, backup_log_files = backup_log_files,
  342. handle = handle, log_info_on_failure=log_info_on_failure)
  343. # Next run_file() invocations should always append to current output
  344. override_output_files = False
  345. if ret['exitcode'] != 0:
  346. break
  347. if not ret: # Something went wrong
  348. raise AgentException("No script has been executed")
  349. # if canceled and not background command
  350. if handle is None:
  351. cancel_reason = self.command_canceled_reason(task_id)
  352. if cancel_reason is not None:
  353. ret['stdout'] += cancel_reason
  354. ret['stderr'] += cancel_reason
  355. with open(tmpoutfile, "a") as f:
  356. f.write(cancel_reason)
  357. with open(tmperrfile, "a") as f:
  358. f.write(cancel_reason)
  359. except Exception, e: # We do not want to let agent fail completely
  360. exc_type, exc_obj, exc_tb = sys.exc_info()
  361. message = "Caught an exception while executing "\
  362. "custom service command: {0}: {1}; {2}".format(exc_type, exc_obj, str(e))
  363. logger.exception(message)
  364. ret = {
  365. 'stdout' : message,
  366. 'stderr' : message,
  367. 'structuredOut' : '{}',
  368. 'exitcode': 1,
  369. }
  370. return ret
  371. def command_canceled_reason(self, task_id):
  372. with self.commands_in_progress_lock:
  373. if self.commands_in_progress.has_key(task_id):#Background command do not push in this collection (TODO)
  374. logger.debug('Pop with taskId %s' % task_id)
  375. pid = self.commands_in_progress.pop(task_id)
  376. if not isinstance(pid, int):
  377. reason = pid
  378. if reason:
  379. return "\nCommand aborted. Reason: '{0}'".format(reason)
  380. else:
  381. return "\nCommand aborted."
  382. return None
  383. def requestComponentStatus(self, command):
  384. """
  385. Component status is determined by exit code, returned by runCommand().
  386. Exit code 0 means that component is running and any other exit code means that
  387. component is not running
  388. """
  389. override_output_files=True # by default, we override status command output
  390. if logger.level == logging.DEBUG:
  391. override_output_files = False
  392. res = self.runCommand(command, self.status_commands_stdout,
  393. self.status_commands_stderr, self.COMMAND_NAME_STATUS,
  394. override_output_files=override_output_files)
  395. return res
  396. def requestComponentSecurityState(self, command):
  397. """
  398. Determines the current security state of the component
  399. A command will be issued to trigger the security_status check and the result of this check will
  400. returned to the caller. If the component lifecycle script has no security_status method the
  401. check will return non zero exit code and "UNKNOWN" will be returned.
  402. """
  403. override_output_files=True # by default, we override status command output
  404. if logger.level == logging.DEBUG:
  405. override_output_files = False
  406. security_check_res = self.runCommand(command, self.status_commands_stdout,
  407. self.status_commands_stderr, self.COMMAND_NAME_SECURITY_STATUS,
  408. override_output_files=override_output_files)
  409. result = 'UNKNOWN'
  410. if security_check_res is None:
  411. logger.warn("The return value of the security_status check was empty, the security status is unknown")
  412. elif 'exitcode' not in security_check_res:
  413. logger.warn("Missing 'exitcode' value from the security_status check result, the security status is unknown")
  414. elif security_check_res['exitcode'] != 0:
  415. logger.debug("The 'exitcode' value from the security_status check result indicated the check routine failed to properly execute, the security status is unknown")
  416. elif 'structuredOut' not in security_check_res:
  417. logger.warn("Missing 'structuredOut' value from the security_status check result, the security status is unknown")
  418. elif 'securityState' not in security_check_res['structuredOut']:
  419. logger.warn("Missing 'securityState' value from the security_status check structuredOut data set, the security status is unknown")
  420. else:
  421. result = security_check_res['structuredOut']['securityState']
  422. return result
  423. def resolve_script_path(self, base_dir, script):
  424. """
  425. Encapsulates logic of script location determination.
  426. """
  427. path = os.path.join(base_dir, script)
  428. if not os.path.exists(path):
  429. message = "Script {0} does not exist".format(path)
  430. raise AgentException(message)
  431. return path
  432. def resolve_hook_script_path(self, stack_hooks_dir, prefix, command_name, script_type):
  433. """
  434. Returns a tuple(path to hook script, hook base dir) according to string prefix
  435. and command name. If script does not exist, returns None
  436. """
  437. if not stack_hooks_dir:
  438. return None
  439. hook_dir = "{0}-{1}".format(prefix, command_name)
  440. hook_base_dir = os.path.join(stack_hooks_dir, hook_dir)
  441. hook_script_path = os.path.join(hook_base_dir, "scripts", "hook.py")
  442. if not os.path.isfile(hook_script_path):
  443. logger.debug("Hook script {0} not found, skipping".format(hook_script_path))
  444. return None
  445. return hook_script_path, hook_base_dir
  446. def dump_command_to_json(self, command, retry=False):
  447. """
  448. Converts command to json file and returns file path
  449. """
  450. # Perform few modifications to stay compatible with the way in which
  451. public_fqdn = self.public_fqdn
  452. command['public_hostname'] = public_fqdn
  453. # Add cache dir to make it visible for commands
  454. command["hostLevelParams"]["agentCacheDir"] = self.config.get('agent', 'cache_dir')
  455. command["agentConfigParams"] = {"agent": {"parallel_execution": self.config.get_parallel_exec_option()}}
  456. # Now, dump the json file
  457. command_type = command['commandType']
  458. from ActionQueue import ActionQueue # To avoid cyclic dependency
  459. if command_type == ActionQueue.STATUS_COMMAND:
  460. # These files are frequently created, that's why we don't
  461. # store them all, but only the latest one
  462. file_path = os.path.join(self.tmp_dir, "status_command.json")
  463. else:
  464. task_id = command['taskId']
  465. if 'clusterHostInfo' in command and command['clusterHostInfo'] and not retry:
  466. command['clusterHostInfo'] = self.decompressClusterHostInfo(command['clusterHostInfo'])
  467. file_path = os.path.join(self.tmp_dir, "command-{0}.json".format(task_id))
  468. if command_type == ActionQueue.AUTO_EXECUTION_COMMAND:
  469. file_path = os.path.join(self.tmp_dir, "auto_command-{0}.json".format(task_id))
  470. # Json may contain passwords, that's why we need proper permissions
  471. if os.path.isfile(file_path):
  472. os.unlink(file_path)
  473. with os.fdopen(os.open(file_path, os.O_WRONLY | os.O_CREAT,
  474. 0600), 'w') as f:
  475. content = json.dumps(command, sort_keys = False, indent = 4)
  476. f.write(content)
  477. return file_path
  478. def decompressClusterHostInfo(self, clusterHostInfo):
  479. info = clusterHostInfo.copy()
  480. #Pop info not related to host roles
  481. hostsList = info.pop(self.HOSTS_LIST_KEY)
  482. pingPorts = info.pop(self.PING_PORTS_KEY)
  483. racks = info.pop(self.RACKS_KEY)
  484. ipv4_addresses = info.pop(self.IPV4_ADDRESSES_KEY)
  485. ambariServerHost = info.pop(self.AMBARI_SERVER_HOST)
  486. ambariServerPort = info.pop(self.AMBARI_SERVER_PORT)
  487. ambariServerUseSsl = info.pop(self.AMBARI_SERVER_USE_SSL)
  488. decompressedMap = {}
  489. for k,v in info.items():
  490. # Convert from 1-3,5,6-8 to [1,2,3,5,6,7,8]
  491. indexes = self.convertRangeToList(v)
  492. # Convert from [1,2,3,5,6,7,8] to [host1,host2,host3...]
  493. decompressedMap[k] = [hostsList[i] for i in indexes]
  494. #Convert from ['1:0-2,4', '42:3,5-7'] to [1,1,1,42,1,42,42,42]
  495. pingPorts = self.convertMappedRangeToList(pingPorts)
  496. racks = self.convertMappedRangeToList(racks)
  497. ipv4_addresses = self.convertMappedRangeToList(ipv4_addresses)
  498. #Convert all elements to str
  499. pingPorts = map(str, pingPorts)
  500. #Add ping ports to result
  501. decompressedMap[self.PING_PORTS_KEY] = pingPorts
  502. #Add hosts list to result
  503. decompressedMap[self.HOSTS_LIST_KEY] = hostsList
  504. #Add racks list to result
  505. decompressedMap[self.RACKS_KEY] = racks
  506. #Add ips list to result
  507. decompressedMap[self.IPV4_ADDRESSES_KEY] = ipv4_addresses
  508. #Add ambari-server properties to result
  509. decompressedMap[self.AMBARI_SERVER_HOST] = ambariServerHost
  510. decompressedMap[self.AMBARI_SERVER_PORT] = ambariServerPort
  511. decompressedMap[self.AMBARI_SERVER_USE_SSL] = ambariServerUseSsl
  512. return decompressedMap
  513. # Converts from 1-3,5,6-8 to [1,2,3,5,6,7,8]
  514. def convertRangeToList(self, list):
  515. resultList = []
  516. for i in list:
  517. ranges = i.split(',')
  518. for r in ranges:
  519. rangeBounds = r.split('-')
  520. if len(rangeBounds) == 2:
  521. if not rangeBounds[0] or not rangeBounds[1]:
  522. raise AgentException("Broken data in given range, expected - ""m-n"" or ""m"", got : " + str(r))
  523. resultList.extend(range(int(rangeBounds[0]), int(rangeBounds[1]) + 1))
  524. elif len(rangeBounds) == 1:
  525. resultList.append((int(rangeBounds[0])))
  526. else:
  527. raise AgentException("Broken data in given range, expected - ""m-n"" or ""m"", got : " + str(r))
  528. return resultList
  529. #Converts from ['1:0-2,4', '42:3,5-7'] to [1,1,1,42,1,42,42,42]
  530. def convertMappedRangeToList(self, list):
  531. resultDict = {}
  532. for i in list:
  533. valueToRanges = i.split(":")
  534. if len(valueToRanges) <> 2:
  535. raise AgentException("Broken data in given value to range, expected format - ""value:m-n"", got - " + str(i))
  536. value = valueToRanges[0]
  537. rangesToken = valueToRanges[1]
  538. for r in rangesToken.split(','):
  539. rangeIndexes = r.split('-')
  540. if len(rangeIndexes) == 2:
  541. if not rangeIndexes[0] or not rangeIndexes[1]:
  542. raise AgentException("Broken data in given value to range, expected format - ""value:m-n"", got - " + str(r))
  543. start = int(rangeIndexes[0])
  544. end = int(rangeIndexes[1])
  545. for k in range(start, end + 1):
  546. resultDict[k] = value if not value.isdigit() else int(value)
  547. elif len(rangeIndexes) == 1:
  548. index = int(rangeIndexes[0])
  549. resultDict[index] = value if not value.isdigit() else int(value)
  550. resultList = dict(sorted(resultDict.items())).values()
  551. return resultList