submission2.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import errno
  18. import logging
  19. import os
  20. import time
  21. from django.utils.translation import ugettext as _
  22. from desktop.lib.exceptions_renderable import PopupException
  23. from desktop.lib.i18n import smart_str
  24. from hadoop import cluster
  25. from hadoop.fs.hadoopfs import Hdfs
  26. from liboozie.oozie_api import get_oozie
  27. from liboozie.conf import REMOTE_DEPLOYMENT_DIR
  28. from jobsub.parameterization import find_variables
  29. from liboozie.credentials import Credentials
  30. LOG = logging.getLogger(__name__)
  31. class Submission(object):
  32. """
  33. Represents one unique Oozie submission.
  34. Actions are:
  35. - submit
  36. - rerun
  37. """
  38. def __init__(self, user, job=None, fs=None, jt=None, properties=None, oozie_id=None):
  39. self.job = job
  40. self.user = user
  41. self.fs = fs
  42. self.jt = jt # Deprecated with YARN, we now use logical names only for RM
  43. self.oozie_id = oozie_id
  44. self.api = get_oozie(self.user)
  45. if properties is not None:
  46. self.properties = properties
  47. else:
  48. self.properties = {}
  49. def __str__(self):
  50. if self.oozie_id:
  51. res = "Submission for job '%s'." % (self.oozie_id,)
  52. else:
  53. res = "Submission for job '%s' (id %s, owner %s)." % (self.job.name, self.job.id, self.user)
  54. if self.oozie_id:
  55. res += " -- " + self.oozie_id
  56. return res
  57. def run(self, deployment_dir=None):
  58. """
  59. Take care of all the actions of submitting a Oozie workflow.
  60. Returns the oozie job id if all goes well.
  61. """
  62. if self.oozie_id is not None:
  63. raise Exception(_("Submission already submitted (Oozie job id %s)") % (self.oozie_id,))
  64. jt_address = cluster.get_cluster_addr_for_job_submission()
  65. if deployment_dir is None:
  66. self._update_properties(jt_address) # Needed as we need to set some properties like Credentials before
  67. deployment_dir = self.deploy()
  68. self._update_properties(jt_address, deployment_dir)
  69. self.oozie_id = self.api.submit_job(self.properties)
  70. LOG.info("Submitted: %s" % (self,))
  71. if self._is_workflow():
  72. self.api.job_control(self.oozie_id, 'start')
  73. LOG.info("Started: %s" % (self,))
  74. return self.oozie_id
  75. def rerun(self, deployment_dir, fail_nodes=None, skip_nodes=None):
  76. jt_address = cluster.get_cluster_addr_for_job_submission()
  77. self._update_properties(jt_address, deployment_dir)
  78. self.properties.update({'oozie.wf.application.path': deployment_dir})
  79. if fail_nodes:
  80. self.properties.update({'oozie.wf.rerun.failnodes': fail_nodes})
  81. elif not skip_nodes:
  82. self.properties.update({'oozie.wf.rerun.failnodes': 'false'}) # Case empty 'skip_nodes' list
  83. else:
  84. self.properties.update({'oozie.wf.rerun.skip.nodes': skip_nodes})
  85. self.api.rerun(self.oozie_id, properties=self.properties)
  86. LOG.info("Rerun: %s" % (self,))
  87. return self.oozie_id
  88. def rerun_coord(self, deployment_dir, params):
  89. jt_address = cluster.get_cluster_addr_for_job_submission()
  90. self._update_properties(jt_address, deployment_dir)
  91. self.properties.update({'oozie.coord.application.path': deployment_dir})
  92. self.api.job_control(self.oozie_id, action='coord-rerun', properties=self.properties, parameters=params)
  93. LOG.info("Rerun: %s" % (self,))
  94. return self.oozie_id
  95. def rerun_bundle(self, deployment_dir, params):
  96. jt_address = cluster.get_cluster_addr_for_job_submission()
  97. self._update_properties(jt_address, deployment_dir)
  98. self.properties.update({'oozie.bundle.application.path': deployment_dir})
  99. self.api.job_control(self.oozie_id, action='bundle-rerun', properties=self.properties, parameters=params)
  100. LOG.info("Rerun: %s" % (self,))
  101. return self.oozie_id
  102. def deploy(self):
  103. try:
  104. deployment_dir = self._create_deployment_dir()
  105. except Exception, ex:
  106. msg = _("Failed to create deployment directory: %s" % ex)
  107. LOG.exception(msg)
  108. raise PopupException(message=msg, detail=str(ex))
  109. oozie_xml = self.job.to_xml(self.properties)
  110. self._do_as(self.user.username , self._copy_files, deployment_dir, oozie_xml)
  111. if hasattr(self.job, 'actions'):
  112. for action in self.job.actions:
  113. # Make sure XML is there
  114. # Don't support shared sub-worfklow, ore more than one level sub-workflow
  115. if action.data['type'] == 'subworkflow':
  116. workflow = Workflow(document=Document2.objects.get(uuid=action.data['properties']['workflow']))
  117. sub_deploy = Submission(self.user, workflow, self.fs, self.jt, self.properties)
  118. sub_deploy.deploy()
  119. return deployment_dir
  120. def get_external_parameters(self, application_path):
  121. """From XML and job.properties HDFS files"""
  122. deployment_dir = os.path.dirname(application_path)
  123. xml = self.fs.do_as_user(self.user, self.fs.read, application_path, 0, 1 * 1024**2)
  124. properties_file = deployment_dir + '/job.properties'
  125. if self.fs.do_as_user(self.user, self.fs.exists, properties_file):
  126. properties = self.fs.do_as_user(self.user, self.fs.read, properties_file, 0, 1 * 1024**2)
  127. else:
  128. properties = None
  129. return self._get_external_parameters(xml, properties)
  130. def _get_external_parameters(self, xml, properties=None):
  131. from oozie.models import DATASET_FREQUENCY
  132. parameters = dict([(var, '') for var in find_variables(xml, include_named=False) if not self._is_coordinator() or var not in DATASET_FREQUENCY])
  133. if properties:
  134. parameters.update(dict([line.strip().split('=')
  135. for line in properties.split('\n') if not line.startswith('#') and len(line.strip().split('=')) == 2]))
  136. return parameters
  137. def _update_properties(self, jobtracker_addr, deployment_dir=None):
  138. LOG.info('Using FS %s and JT %s' % (self.fs, self.jt))
  139. if self.jt and self.jt.logical_name:
  140. jobtracker_addr = self.jt.logical_name
  141. if self.fs.logical_name:
  142. fs_defaultfs = self.fs.logical_name
  143. else:
  144. fs_defaultfs = self.fs.fs_defaultfs
  145. self.properties.update({
  146. 'jobTracker': jobtracker_addr,
  147. 'nameNode': fs_defaultfs,
  148. })
  149. if self.job and deployment_dir:
  150. self.properties.update({
  151. self.job.PROPERTY_APP_PATH: self.fs.get_hdfs_path(deployment_dir),
  152. self.job.HUE_ID: self.job.id
  153. })
  154. # Generate credentials when using security
  155. if self.api.security_enabled:
  156. credentials = Credentials()
  157. credentials.fetch(self.api)
  158. self.properties['credentials'] = credentials.get_properties()
  159. def _create_deployment_dir(self):
  160. """
  161. Return the job deployment directory in HDFS, creating it if necessary.
  162. The actual deployment dir should be 0711 owned by the user
  163. """
  164. # Automatic setup of the required directories if needed
  165. create_directories(self.fs)
  166. # Case of a shared job
  167. if self.user != self.job.document.owner:
  168. path = REMOTE_DEPLOYMENT_DIR.get().replace('$USER', self.user.username).replace('$TIME', str(time.time())).replace('$JOBID', str(self.job.id))
  169. # Shared coords or bundles might not have any existing workspaces
  170. if self.fs.exists(self.job.deployment_dir):
  171. self.fs.copy_remote_dir(self.job.deployment_dir, path, owner=self.user, dir_mode=0711)
  172. else:
  173. self._create_dir(path)
  174. else:
  175. path = self.job.deployment_dir
  176. self._create_dir(path)
  177. return path
  178. def _create_dir(self, path, perms=0711):
  179. """
  180. Return the directory in HDFS, creating it if necessary.
  181. """
  182. try:
  183. statbuf = self.fs.stats(path)
  184. if not statbuf.isDir:
  185. msg = _("Path is not a directory: %s.") % (path,)
  186. LOG.error(msg)
  187. raise Exception(msg)
  188. except IOError, ex:
  189. if ex.errno != errno.ENOENT:
  190. msg = _("Error accessing directory '%s': %s.") % (path, ex)
  191. LOG.exception(msg)
  192. raise IOError(ex.errno, msg)
  193. if not self.fs.exists(path):
  194. self._do_as(self.user.username, self.fs.mkdir, path, perms)
  195. self._do_as(self.user.username, self.fs.chmod, path, perms)
  196. return path
  197. def _copy_files(self, deployment_dir, oozie_xml):
  198. """
  199. Copy XML and the jar_path files from Java or MR actions to the deployment directory.
  200. This should run as the workflow user.
  201. """
  202. xml_path = self.fs.join(deployment_dir, self.job.XML_FILE_NAME)
  203. self.fs.create(xml_path, overwrite=True, permission=0644, data=smart_str(oozie_xml))
  204. LOG.debug("Created %s" % (xml_path,))
  205. # List jar files
  206. files = []
  207. lib_path = self.fs.join(deployment_dir, 'lib')
  208. if hasattr(self.job, 'node_list'):
  209. for node in self.job.node_list:
  210. if hasattr(node, 'jar_path') and not node.jar_path.startswith(lib_path):
  211. files.append(node.jar_path)
  212. # Copy the jar files to the workspace lib
  213. if files:
  214. for jar_file in files:
  215. LOG.debug("Updating %s" % jar_file)
  216. jar_lib_path = self.fs.join(lib_path, self.fs.basename(jar_file))
  217. # Refresh if needed
  218. if self.fs.exists(jar_lib_path):
  219. stat_src = self.fs.stats(jar_file)
  220. stat_dest = self.fs.stats(jar_lib_path)
  221. if stat_src.fileId != stat_dest.fileId:
  222. self.fs.remove(jar_lib_path, skip_trash=True)
  223. self.fs.copyfile(jar_file, jar_lib_path)
  224. def _do_as(self, username, fn, *args, **kwargs):
  225. prev_user = self.fs.user
  226. try:
  227. self.fs.setuser(username)
  228. return fn(*args, **kwargs)
  229. finally:
  230. self.fs.setuser(prev_user)
  231. def remove_deployment_dir(self):
  232. """Delete the workflow deployment directory."""
  233. try:
  234. path = self.job.deployment_dir
  235. if self._do_as(self.user.username , self.fs.exists, path):
  236. self._do_as(self.user.username , self.fs.rmtree, path)
  237. except Exception, ex:
  238. LOG.warn("Failed to clean up workflow deployment directory for "
  239. "%s (owner %s). Caused by: %s",
  240. self.job.name, self.user, ex)
  241. def _is_workflow(self):
  242. from oozie.models2 import Workflow
  243. return Workflow.PROPERTY_APP_PATH in self.properties
  244. def _is_coordinator(self):
  245. from oozie.models2 import Coordinator
  246. return Coordinator.PROPERTY_APP_PATH in self.properties
  247. def create_directories(fs, directory_list=[]):
  248. # If needed, create the remote home, deployment and data directories
  249. directories = [REMOTE_DEPLOYMENT_DIR.get()] + directory_list
  250. for directory in directories:
  251. if not fs.do_as_user(fs.DEFAULT_USER, fs.exists, directory):
  252. remote_home_dir = Hdfs.join('/user', fs.DEFAULT_USER)
  253. if directory.startswith(remote_home_dir):
  254. # Home is 755
  255. fs.do_as_user(fs.DEFAULT_USER, fs.create_home_dir, remote_home_dir)
  256. # Shared by all the users
  257. fs.do_as_user(fs.DEFAULT_USER, fs.mkdir, directory, 01777)
  258. fs.do_as_user(fs.DEFAULT_USER, fs.chmod, directory, 01777) # To remove after https://issues.apache.org/jira/browse/HDFS-3491