submission2.py 12 KB

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