submit.py 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  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. """
  18. Handle design submission.
  19. """
  20. import errno
  21. import logging
  22. from desktop.lib import django_mako
  23. from desktop.lib.django_util import PopupException
  24. import hadoop.cluster
  25. from hadoop.fs.hadoopfs import Hdfs
  26. from jobsub import conf, models
  27. from jobsub.oozie_lib.oozie_api import get_oozie
  28. LOG = logging.getLogger(__name__)
  29. class Submission(object):
  30. """Represents one submission"""
  31. def __init__(self, design_obj, fs):
  32. self._design_obj = design_obj
  33. self._username = design_obj.owner.username
  34. self._action = design_obj.get_root_action()
  35. self._fs = fs
  36. self._job_id = None # The oozie workflow instance id
  37. def __unicode__(self):
  38. res = "Submission for job design '%s' (id %s, owner %s)" % \
  39. (self._design_obj.name, self._design_obj.id, self._username)
  40. if self.job_id:
  41. res += " -- " + self.job_id
  42. return res
  43. @property
  44. def job_id(self):
  45. return self._job_id
  46. def _do_as(self, username, fn, *args, **kwargs):
  47. curr_user = self._fs.setuser(username)
  48. try:
  49. fn(*args, **kwargs)
  50. finally:
  51. self._fs.setuser(curr_user)
  52. def run(self):
  53. """
  54. Take care of all the actions of submitting a workflow/design.
  55. Returns the oozie job id if all goes well.
  56. """
  57. if self.job_id is not None:
  58. raise Exception("Job design already submitted (Oozie job id %s)" % (self.job_id,))
  59. fs_defaultfs = self._fs.fs_defaultfs
  60. jobtracker = hadoop.cluster.get_cluster_addr_for_job_submission()
  61. try:
  62. wf_dir = self._get_and_create_deployment_dir()
  63. except Exception, ex:
  64. LOG.exception("Failed to access deployment directory")
  65. raise PopupException(message="Failed to access deployment directory",
  66. detail=str(ex))
  67. wf_xml = self._generate_workflow_xml(fs_defaultfs)
  68. self._do_as(self._username, self._copy_files, wf_dir, wf_xml)
  69. LOG.info("Prepared deployment directory at '%s' for %s" % (wf_dir, self))
  70. LOG.info("Submitting design id %s to %s as `%s'" % (self._design_obj.id, jobtracker, self._username))
  71. try:
  72. prev = get_oozie().setuser(self._username)
  73. self._job_id = get_oozie().submit_workflow(
  74. self._fs.get_hdfs_path(wf_dir),
  75. properties=self._get_properties(jobtracker))
  76. LOG.info("Submitted: %s" % (self,))
  77. # Now we need to run it
  78. get_oozie().job_control(self.job_id, 'start')
  79. LOG.info("Started: %s" % (self,))
  80. finally:
  81. get_oozie().setuser(prev)
  82. return self.job_id
  83. def _get_properties(self, jobtracker_addr):
  84. res = { 'jobTracker': jobtracker_addr }
  85. if self._design_obj.get_root_action().action_type == \
  86. models.OozieStreamingAction.ACTION_TYPE:
  87. res['oozie.use.system.libpath'] = 'true'
  88. return res
  89. def _copy_files(self, wf_dir, wf_xml):
  90. """
  91. Copy the files over to the deployment directory. This should run as the
  92. design owner.
  93. """
  94. xml_path = self._fs.join(wf_dir, 'workflow.xml')
  95. self._fs.create(xml_path, overwrite=True, permission=0644, data=wf_xml)
  96. LOG.debug("Created %s" % (xml_path,))
  97. # Copy the jar over
  98. if self._action.action_type in (models.OozieMapreduceAction.ACTION_TYPE,
  99. models.OozieJavaAction.ACTION_TYPE):
  100. lib_path = self._fs.join(wf_dir, 'lib')
  101. if self._fs.exists(lib_path):
  102. LOG.debug("Cleaning up old %s" % (lib_path,))
  103. self._fs.rmtree(lib_path)
  104. self._fs.mkdir(lib_path, 0755)
  105. LOG.debug("Created %s" % (lib_path,))
  106. jar = self._action.jar_path
  107. self._fs.copyfile(jar, self._fs.join(lib_path, self._fs.basename(jar)))
  108. def _generate_workflow_xml(self, namenode):
  109. """Return a string that is the workflow.xml of this workflow"""
  110. action_type = self._design_obj.root_action.action_type
  111. data = {
  112. 'design': self._design_obj,
  113. 'nameNode': namenode,
  114. }
  115. if action_type == models.OozieStreamingAction.ACTION_TYPE:
  116. tmpl = "workflow-streaming.xml.mako"
  117. elif action_type == models.OozieMapreduceAction.ACTION_TYPE:
  118. tmpl = "workflow-mapreduce.xml.mako"
  119. elif action_type == models.OozieJavaAction.ACTION_TYPE:
  120. tmpl = "workflow-java.xml.mako"
  121. return django_mako.render_to_string(tmpl, data)
  122. def _get_and_create_deployment_dir(self):
  123. """
  124. Return the workflow deployment directory in HDFS,
  125. creating it if necessary.
  126. May raise Exception.
  127. """
  128. path = self._get_deployment_dir()
  129. try:
  130. statbuf = self._fs.stats(path)
  131. if not statbuf.isDir:
  132. msg = "Workflow deployment path is not a directory: %s" % (path,)
  133. LOG.error(msg)
  134. raise Exception(msg)
  135. return path
  136. except IOError, ex:
  137. if ex.errno != errno.ENOENT:
  138. msg = "Error accessing workflow directory '%s': %s" % (path, ex)
  139. LOG.exception(msg)
  140. raise IOError(ex.errno, msg)
  141. self._create_deployment_dir(path)
  142. return path
  143. def _create_deployment_dir(self, path):
  144. # Make sure the root data dir exists
  145. self.create_data_dir(self._fs)
  146. # The actual deployment dir should be 0711 owned by the user
  147. self._do_as(self._username, self._fs.mkdir, path, 0711)
  148. @classmethod
  149. def create_data_dir(cls, fs):
  150. # If needed, create the remote home and data directories
  151. remote_data_dir = conf.REMOTE_DATA_DIR.get()
  152. user = fs.user
  153. try:
  154. fs.setuser(fs.DEFAULT_USER)
  155. if not fs.exists(remote_data_dir):
  156. remote_home_dir = Hdfs.join('/user', fs.user)
  157. if remote_data_dir.startswith(remote_home_dir):
  158. # Home is 755
  159. fs.create_home_dir(remote_home_dir)
  160. # Shared by all the users
  161. fs.mkdir(remote_data_dir, 01777)
  162. finally:
  163. fs.setuser(user)
  164. return remote_data_dir
  165. def _get_deployment_dir(self):
  166. """Return the workflow deployment directory"""
  167. if self._fs is None:
  168. raise PopupException("Failed to obtain HDFS reference. "
  169. "Please check your configuration.")
  170. # We could have collision with usernames. But there's no good separator.
  171. # Hope people don't create crazy usernames.
  172. return self._fs.join(conf.REMOTE_DATA_DIR.get(),
  173. "_%s_-design-%s" % (self._username, self._design_obj.id))
  174. def remove_deployment_dir(self):
  175. """Delete the workflow deployment directory. Does not throw."""
  176. try:
  177. path = self._get_deployment_dir()
  178. if self._do_as(self._username, self._fs.exists, path):
  179. self._do_as(self._username, self._fs.rmtree, path)
  180. except Exception, ex:
  181. LOG.warn("Failed to clean up workflow deployment directory for "
  182. "%s (owner %s). Caused by: %s",
  183. self._design_obj.name, self._design_obj.owner.username, ex)