api.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244
  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 json
  18. import logging
  19. import time
  20. from django.core.urlresolvers import reverse
  21. from django.utils.translation import ugettext as _
  22. from desktop.lib.i18n import smart_str
  23. from desktop.lib.view_util import format_duration_in_millis
  24. from liboozie.oozie_api import get_oozie
  25. from oozie.models import Workflow, Pig
  26. from oozie.views.api import get_log as get_workflow_logs
  27. from oozie.views.editor import _submit_workflow
  28. LOG = logging.getLogger(__name__)
  29. def get(fs, jt, user):
  30. return OozieApi(fs, jt, user)
  31. class OozieApi(object):
  32. """
  33. Oozie submission.
  34. """
  35. WORKFLOW_NAME = 'pig-app-hue-script'
  36. LOG_START_PATTERN = '(Pig script \[(?:[\w.-]+)\] content:.+)'
  37. LOG_END_PATTERN = '(<<< Invocation of Pig command completed <<<|' \
  38. '<<< Invocation of Main class completed <<<|' \
  39. '<<< Invocation of Pig command completed <<<|' \
  40. '<<< Invocation of Main class completed <<<)'
  41. MAX_DASHBOARD_JOBS = 100
  42. def __init__(self, fs, jt, user):
  43. self.oozie_api = get_oozie(user)
  44. self.fs = fs
  45. self.jt = jt
  46. self.user = user
  47. def submit(self, pig_script, params):
  48. workflow = None
  49. try:
  50. workflow = self._create_workflow(pig_script, params)
  51. mapping = dict([(param['name'], param['value']) for param in workflow.get_parameters()])
  52. oozie_wf = _submit_workflow(self.user, self.fs, self.jt, workflow, mapping)
  53. finally:
  54. if workflow:
  55. workflow.delete(skip_trash=True)
  56. return oozie_wf
  57. def _create_workflow(self, pig_script, params):
  58. workflow = Workflow.objects.new_workflow(self.user)
  59. workflow.schema_version = 'uri:oozie:workflow:0.5'
  60. workflow.name = OozieApi.WORKFLOW_NAME
  61. workflow.is_history = True
  62. if pig_script.use_hcatalog:
  63. workflow.add_parameter("oozie.action.sharelib.for.pig", "pig,hcatalog")
  64. workflow.save()
  65. Workflow.objects.initialize(workflow, self.fs)
  66. script_path = workflow.deployment_dir + '/script.pig'
  67. if self.fs: # For testing, difficult to mock
  68. self.fs.do_as_user(self.user.username, self.fs.create, script_path, data=smart_str(pig_script.dict['script']))
  69. files = []
  70. archives = []
  71. popup_params = json.loads(params)
  72. popup_params_names = [param['name'] for param in popup_params]
  73. pig_params = self._build_parameters(popup_params)
  74. if pig_script.isV2:
  75. pig_params += [{"type": "argument", "value": param} for param in pig_script.dict['parameters']]
  76. job_properties = [{"name": prop.split('=', 1)[0], "value": prop.split('=', 1)[1]} for prop in pig_script.dict['hadoopProperties']]
  77. for resource in pig_script.dict['resources']:
  78. if resource.endswith('.zip') or resource.endswith('.tgz') or resource.endswith('.tar') or resource.endswith('.gz'):
  79. archives.append({"dummy": "", "name": resource})
  80. else:
  81. files.append(resource)
  82. else:
  83. script_params = [param for param in pig_script.dict['parameters'] if param['name'] not in popup_params_names]
  84. pig_params += self._build_parameters(script_params)
  85. job_properties = [{"name": prop['name'], "value": prop['value']} for prop in pig_script.dict['hadoopProperties']]
  86. for resource in pig_script.dict['resources']:
  87. if resource['type'] == 'file':
  88. files.append(resource['value'])
  89. if resource['type'] == 'archive':
  90. archives.append({"dummy": "", "name": resource['value']})
  91. action = Pig.objects.create(
  92. name='pig-5760',
  93. script_path=script_path,
  94. workflow=workflow,
  95. node_type='pig',
  96. params=json.dumps(pig_params),
  97. files=json.dumps(files),
  98. archives=json.dumps(archives),
  99. job_properties=json.dumps(job_properties)
  100. )
  101. credentials = []
  102. if pig_script.use_hcatalog and self.oozie_api.security_enabled:
  103. credentials.append({'name': 'hcat', 'value': True})
  104. if pig_script.use_hbase and self.oozie_api.security_enabled:
  105. credentials.append({'name': 'hbase', 'value': True})
  106. if credentials:
  107. action.credentials = credentials # Note, action.credentials is a @setter here
  108. action.save()
  109. action.add_node(workflow.end)
  110. start_link = workflow.start.get_link()
  111. start_link.child = action
  112. start_link.save()
  113. return workflow
  114. def _build_parameters(self, params):
  115. pig_params = []
  116. for param in params:
  117. if param['name'].startswith('-'):
  118. pig_params.append({"type": "argument", "value": "%(name)s" % param})
  119. if param['value']:
  120. pig_params.append({"type": "argument", "value": "%(value)s" % param})
  121. else:
  122. # Simpler way and backward compatibility for parameters
  123. pig_params.append({"type": "argument", "value": "-param"})
  124. pig_params.append({"type": "argument", "value": "%(name)s=%(value)s" % param})
  125. return pig_params
  126. def stop(self, job_id):
  127. return self.oozie_api.job_control(job_id, 'kill')
  128. def get_jobs(self):
  129. kwargs = {'cnt': OozieApi.MAX_DASHBOARD_JOBS,}
  130. kwargs['filters'] = [
  131. ('user', self.user.username),
  132. ('name', OozieApi.WORKFLOW_NAME)
  133. ]
  134. return self.oozie_api.get_workflows(**kwargs).jobs
  135. def get_log(self, request, oozie_workflow, make_links=True):
  136. return get_workflow_logs(request, oozie_workflow, make_links=make_links, log_start_pattern=self.LOG_START_PATTERN,
  137. log_end_pattern=self.LOG_END_PATTERN)
  138. def massaged_jobs_for_json(self, request, oozie_jobs, hue_jobs):
  139. jobs = []
  140. hue_jobs = dict([(script.dict.get('job_id'), script) for script in hue_jobs if script.dict.get('job_id')])
  141. for job in oozie_jobs:
  142. if job.is_running():
  143. job = self.oozie_api.get_job(job.id)
  144. get_copy = request.GET.copy() # Hacky, would need to refactor JobBrowser get logs
  145. get_copy['format'] = 'python'
  146. request.GET = get_copy
  147. try:
  148. logs, workflow_action, is_really_done = self.get_log(request, job)
  149. progress = workflow_action[0]['progress']
  150. except:
  151. LOG.exception('failed to get progress')
  152. progress = 0
  153. else:
  154. progress = 100
  155. hue_pig = hue_jobs.get(job.id) and hue_jobs.get(job.id) or None
  156. massaged_job = {
  157. 'id': job.id,
  158. 'lastModTime': hasattr(job, 'lastModTime') and job.lastModTime and format_time(job.lastModTime) or None,
  159. 'kickoffTime': hasattr(job, 'kickoffTime') and job.kickoffTime or None,
  160. 'timeOut': hasattr(job, 'timeOut') and job.timeOut or None,
  161. 'endTime': job.endTime and format_time(job.endTime) or None,
  162. 'status': job.status,
  163. 'isRunning': job.is_running(),
  164. 'duration': job.endTime and job.startTime and format_duration_in_millis(( time.mktime(job.endTime) - time.mktime(job.startTime) ) * 1000) or None,
  165. 'appName': hue_pig and hue_pig.dict['name'] or _('Unsaved script'),
  166. 'scriptId': hue_pig and hue_pig.id or -1,
  167. 'scriptContent': hue_pig and hue_pig.dict['script'] or '',
  168. 'progress': progress,
  169. 'progressPercent': '%d%%' % progress,
  170. 'user': job.user,
  171. 'absoluteUrl': job.get_absolute_url(),
  172. 'canEdit': has_job_edition_permission(job, self.user),
  173. 'killUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'kill'}),
  174. 'watchUrl': reverse('pig:watch', kwargs={'job_id': job.id}) + '?format=python',
  175. 'created': hasattr(job, 'createdTime') and job.createdTime and job.createdTime and ((job.type == 'Bundle' and job.createdTime) or format_time(job.createdTime)),
  176. 'startTime': hasattr(job, 'startTime') and format_time(job.startTime) or None,
  177. 'run': hasattr(job, 'run') and job.run or 0,
  178. 'frequency': hasattr(job, 'frequency') and job.frequency or None,
  179. 'timeUnit': hasattr(job, 'timeUnit') and job.timeUnit or None,
  180. }
  181. jobs.append(massaged_job)
  182. return jobs
  183. def format_time(st_time):
  184. if st_time is None:
  185. return '-'
  186. else:
  187. return time.strftime("%a, %d %b %Y %H:%M:%S", st_time)
  188. def has_job_edition_permission(oozie_job, user):
  189. return user.is_superuser or oozie_job.user == user.username