dashboard.py 47 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200
  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 os
  20. import re
  21. import time
  22. from django.forms.formsets import formset_factory
  23. from django.http import HttpResponse
  24. from django.utils.functional import wraps
  25. from django.utils.translation import ugettext as _
  26. from django.urls import reverse
  27. from django.shortcuts import redirect
  28. from desktop.conf import TIME_ZONE
  29. from desktop.lib import django_mako
  30. from desktop.lib.django_util import JsonResponse, render
  31. from desktop.lib.json_utils import JSONEncoderForHTML
  32. from desktop.lib.exceptions_renderable import PopupException
  33. from desktop.lib.i18n import smart_str, smart_unicode
  34. from desktop.lib.rest.http_client import RestException
  35. from desktop.lib.view_util import format_duration_in_millis
  36. from desktop.log.access import access_warn
  37. from desktop.models import Document, Document2
  38. from hadoop.fs.hadoopfs import Hdfs
  39. from liboozie.credentials import Credentials
  40. from liboozie.oozie_api import get_oozie
  41. from liboozie.submission2 import Submission
  42. from liboozie.utils import catch_unicode_time
  43. from oozie.conf import OOZIE_JOBS_COUNT, ENABLE_CRON_SCHEDULING, ENABLE_V2, ENABLE_OOZIE_BACKEND_FILTERING
  44. from oozie.forms import RerunForm, ParameterForm, RerunCoordForm, RerunBundleForm, UpdateCoordinatorForm
  45. from oozie.models import Workflow as OldWorkflow, Job, utc_datetime_format, Bundle, Coordinator, get_link, History as OldHistory
  46. from oozie.models2 import History, Workflow, WORKFLOW_NODE_PROPERTIES
  47. from oozie.settings import DJANGO_APPS
  48. from oozie.utils import convert_to_server_timezone
  49. def get_history():
  50. if ENABLE_V2.get():
  51. return History
  52. else:
  53. return OldHistory
  54. def get_workflow():
  55. if ENABLE_V2.get():
  56. return Workflow
  57. else:
  58. return OldWorkflow
  59. LOG = logging.getLogger(__name__)
  60. """
  61. Permissions:
  62. A Workflow/Coordinator/Bundle can:
  63. * be accessed only by its owner or a superuser or by a user with 'dashboard_jobs_access' permissions
  64. * be submitted/modified only by its owner or a superuser
  65. Permissions checking happens by calling:
  66. * check_job_access_permission()
  67. * check_job_edition_permission()
  68. """
  69. def _get_workflows(user):
  70. return [{
  71. 'name': workflow.name,
  72. 'owner': workflow.owner.username,
  73. 'value': workflow.uuid,
  74. 'id': workflow.id
  75. } for workflow in [d.content_object for d in Document.objects.get_docs(user, Document2, extra='workflow2')]
  76. ]
  77. def manage_oozie_jobs(request, job_id, action):
  78. if request.method != 'POST':
  79. raise PopupException(_('Use a POST request to manage an Oozie job.'))
  80. job = check_job_access_permission(request, job_id)
  81. check_job_edition_permission(job, request.user)
  82. response = {'status': -1, 'data': ''}
  83. try:
  84. oozie_api = get_oozie(request.user)
  85. params = None
  86. if action == 'change':
  87. pause_time_val = request.POST.get('pause_time')
  88. if request.POST.get('clear_pause_time') == 'true':
  89. pause_time_val = ''
  90. end_time_val = request.POST.get('end_time')
  91. if end_time_val:
  92. end_time_val = convert_to_server_timezone(end_time_val, TIME_ZONE.get())
  93. if pause_time_val:
  94. pause_time_val = convert_to_server_timezone(pause_time_val, TIME_ZONE.get())
  95. params = {'value': 'endtime=%s' % (end_time_val) + ';'
  96. 'pausetime=%s' % (pause_time_val) + ';'
  97. 'concurrency=%s' % (request.POST.get('concurrency'))}
  98. elif action == 'ignore':
  99. oozie_api = get_oozie(request.user, api_version="v2")
  100. params = {
  101. 'type': 'action',
  102. 'scope': ','.join(job.aggreate(request.POST.get('actions').split())),
  103. }
  104. response['data'] = oozie_api.job_control(job_id, action, parameters=params)
  105. response['status'] = 0
  106. if 'notification' in request.POST:
  107. request.info(_(request.POST.get('notification')))
  108. except RestException, ex:
  109. ex_message = ex.message
  110. if ex._headers.get('oozie-error-message'):
  111. ex_message = ex._headers.get('oozie-error-message')
  112. msg = "Error performing %s on Oozie job %s: %s." % (action, job_id, ex_message)
  113. LOG.exception(msg)
  114. response['data'] = _(msg)
  115. return JsonResponse(response)
  116. def bulk_manage_oozie_jobs(request):
  117. if request.method != 'POST':
  118. raise PopupException(_('Use a POST request to manage the Oozie jobs.'))
  119. response = {'status': -1, 'data': ''}
  120. if 'job_ids' in request.POST and 'action' in request.POST:
  121. jobs = request.POST.get('job_ids').split()
  122. response = {'totalRequests': len(jobs), 'totalErrors': 0, 'messages': ''}
  123. oozie_api = get_oozie(request.user)
  124. for job_id in jobs:
  125. job = check_job_access_permission(request, job_id)
  126. check_job_edition_permission(job, request.user)
  127. try:
  128. oozie_api.job_control(job_id, request.POST.get('action'))
  129. except RestException, ex:
  130. LOG.exception("Error performing bulk operation for job_id=%s", job_id)
  131. response['totalErrors'] = response['totalErrors'] + 1
  132. response['messages'] += str(ex)
  133. return JsonResponse(response)
  134. def show_oozie_error(view_func):
  135. def decorate(request, *args, **kwargs):
  136. try:
  137. return view_func(request, *args, **kwargs)
  138. except RestException, ex:
  139. LOG.exception("Error communicating with Oozie in %s", view_func.__name__)
  140. detail = ex._headers.get('oozie-error-message', ex)
  141. if 'Max retries exceeded with url' in str(detail) or 'Connection refused' in str(detail):
  142. detail = _('The Oozie server is not running')
  143. raise PopupException(_('An error occurred with Oozie.'), detail=detail)
  144. return wraps(view_func)(decorate)
  145. @show_oozie_error
  146. def list_oozie_workflows(request):
  147. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(), 'filters': []}
  148. if not has_dashboard_jobs_access(request.user):
  149. kwargs['filters'].append(('user', request.user.username))
  150. oozie_api = get_oozie(request.user)
  151. if request.GET.get('format') == 'json':
  152. just_sla = request.GET.get('justsla') == 'true'
  153. if request.GET.get('startcreatedtime'):
  154. kwargs['filters'].extend([('startcreatedtime', request.GET.get('startcreatedtime'))])
  155. if request.GET.get('text') and ENABLE_OOZIE_BACKEND_FILTERING.get():
  156. kwargs['filters'].extend([('text', request.GET.get('text'))])
  157. if request.GET.get('offset'):
  158. kwargs['offset'] = request.GET.get('offset')
  159. json_jobs = []
  160. total_jobs = 0
  161. if request.GET.getlist('status'):
  162. kwargs['filters'].extend([('status', status) for status in request.GET.getlist('status')])
  163. wf_list = oozie_api.get_workflows(**kwargs)
  164. json_jobs = wf_list.jobs
  165. total_jobs = wf_list.total
  166. if request.GET.get('type') == 'progress':
  167. json_jobs = [oozie_api.get_job(job.id) for job in json_jobs]
  168. response = massaged_oozie_jobs_for_json(json_jobs, request.user, just_sla)
  169. response['total_jobs'] = total_jobs
  170. return JsonResponse(response, encoder=JSONEncoderForHTML)
  171. return render('dashboard/list_oozie_workflows.mako', request, {
  172. 'user': request.user,
  173. 'jobs': [],
  174. 'has_job_edition_permission': has_job_edition_permission,
  175. })
  176. @show_oozie_error
  177. def list_oozie_coordinators(request):
  178. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(), 'filters': []}
  179. if not has_dashboard_jobs_access(request.user):
  180. kwargs['filters'].append(('user', request.user.username))
  181. oozie_api = get_oozie(request.user)
  182. enable_cron_scheduling = ENABLE_CRON_SCHEDULING.get()
  183. if request.GET.get('format') == 'json':
  184. if request.GET.get('offset'):
  185. kwargs['offset'] = request.GET.get('offset')
  186. if request.GET.get('text') and ENABLE_OOZIE_BACKEND_FILTERING.get():
  187. kwargs['filters'].extend([('text', request.GET.get('text'))])
  188. json_jobs = []
  189. total_jobs = 0
  190. if request.GET.getlist('status'):
  191. kwargs['filters'].extend([('status', status) for status in request.GET.getlist('status')])
  192. co_list = oozie_api.get_coordinators(**kwargs)
  193. json_jobs = co_list.jobs
  194. total_jobs = co_list.total
  195. if request.GET.get('type') == 'progress':
  196. json_jobs = [oozie_api.get_coordinator(job.id) for job in json_jobs]
  197. response = massaged_oozie_jobs_for_json(json_jobs, request.user)
  198. response['total_jobs'] = total_jobs
  199. return JsonResponse(response, encoder=JSONEncoderForHTML)
  200. return render('dashboard/list_oozie_coordinators.mako', request, {
  201. 'jobs': [],
  202. 'has_job_edition_permission': has_job_edition_permission,
  203. 'enable_cron_scheduling': enable_cron_scheduling,
  204. })
  205. @show_oozie_error
  206. def list_oozie_bundles(request):
  207. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(), 'filters': []}
  208. if not has_dashboard_jobs_access(request.user):
  209. kwargs['filters'].append(('user', request.user.username))
  210. oozie_api = get_oozie(request.user)
  211. if request.GET.get('format') == 'json':
  212. if request.GET.get('offset'):
  213. kwargs['offset'] = request.GET.get('offset')
  214. if request.GET.get('text') and ENABLE_OOZIE_BACKEND_FILTERING.get():
  215. kwargs['filters'].extend([('text', request.GET.get('text'))])
  216. json_jobs = []
  217. total_jobs = 0
  218. if request.GET.getlist('status'):
  219. kwargs['filters'].extend([('status', status) for status in request.GET.getlist('status')])
  220. bundle_list = oozie_api.get_bundles(**kwargs)
  221. json_jobs = bundle_list.jobs
  222. total_jobs = bundle_list.total
  223. if request.GET.get('type') == 'progress':
  224. json_jobs = [oozie_api.get_coordinator(job.id) for job in json_jobs]
  225. response = massaged_oozie_jobs_for_json(json_jobs, request.user)
  226. response['total_jobs'] = total_jobs
  227. return JsonResponse(response, encoder=JSONEncoderForHTML)
  228. return render('dashboard/list_oozie_bundles.mako', request, {
  229. 'jobs': [],
  230. 'has_job_edition_permission': has_job_edition_permission,
  231. })
  232. @show_oozie_error
  233. def list_oozie_workflow(request, job_id):
  234. oozie_workflow = check_job_access_permission(request, job_id)
  235. oozie_coordinator = None
  236. if request.GET.get('coordinator_job_id'):
  237. oozie_coordinator = check_job_access_permission(request, request.GET.get('coordinator_job_id'))
  238. oozie_bundle = None
  239. if request.GET.get('bundle_job_id'):
  240. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  241. if oozie_coordinator is not None:
  242. setattr(oozie_workflow, 'oozie_coordinator', oozie_coordinator)
  243. if oozie_bundle is not None:
  244. setattr(oozie_workflow, 'oozie_bundle', oozie_bundle)
  245. oozie_parent = oozie_workflow.get_parent_job_id()
  246. if oozie_parent:
  247. oozie_parent = check_job_access_permission(request, oozie_parent)
  248. workflow_data = {}
  249. credentials = None
  250. doc = None
  251. hue_workflow = None
  252. hue_coord = None
  253. workflow_graph = 'MISSING' # default to prevent loading the graph tab for deleted workflows
  254. full_node_list = None
  255. if ENABLE_V2.get():
  256. try:
  257. # To update with the new History document model
  258. hue_coord = get_history().get_coordinator_from_config(oozie_workflow.conf_dict)
  259. hue_workflow = get_history().get_workflow_from_config(oozie_workflow.conf_dict)
  260. # When a workflow is submitted by a coordinator
  261. if not hue_workflow and hue_coord and hue_coord.workflow.document:
  262. hue_workflow = hue_coord.workflow
  263. if hue_coord and hue_coord.workflow and hue_coord.workflow.document: hue_coord.workflow.document.doc.get().can_read_or_exception(request.user)
  264. if hue_workflow: hue_workflow.document.doc.get().can_read_or_exception(request.user)
  265. if hue_workflow:
  266. full_node_list = hue_workflow.nodes
  267. workflow_id = hue_workflow.id
  268. wid = {
  269. 'id': workflow_id
  270. }
  271. doc = Document2.objects.get(type='oozie-workflow2', **wid)
  272. new_workflow = get_workflow()(document=doc)
  273. workflow_data = new_workflow.get_data()
  274. except Exception, e:
  275. LOG.exception("Error generating full page for running workflow %s with exception: %s" % (job_id, e.message))
  276. finally:
  277. workflow_graph = ''
  278. credentials = Credentials()
  279. if not workflow_data.get('layout') or oozie_workflow.conf_dict.get('submit_single_action'):
  280. try:
  281. workflow_data = Workflow.gen_workflow_data_from_xml(request.user, oozie_workflow)
  282. # Hide graph tab when node count > 30
  283. if workflow_data.get('workflow') and len(workflow_data.get('workflow')['nodes']) > 30:
  284. workflow_data = {}
  285. except Exception, e:
  286. LOG.exception('Graph data could not be generated from Workflow %s: %s' % (oozie_workflow.id, e))
  287. else:
  288. history = get_history().cross_reference_submission_history(request.user, job_id)
  289. hue_coord = history and history.get_coordinator() or get_history().get_coordinator_from_config(oozie_workflow.conf_dict)
  290. hue_workflow = (hue_coord and hue_coord.workflow) or (history and history.get_workflow()) or get_history().get_workflow_from_config(oozie_workflow.conf_dict)
  291. if hue_coord and hue_coord.workflow: Job.objects.can_read_or_exception(request, hue_coord.workflow.id)
  292. if hue_workflow: Job.objects.can_read_or_exception(request, hue_workflow.id)
  293. if hue_workflow:
  294. workflow_graph = hue_workflow.gen_status_graph(oozie_workflow)
  295. full_node_list = hue_workflow.node_list
  296. else:
  297. workflow_graph, full_node_list = get_workflow().gen_status_graph_from_xml(request.user, oozie_workflow)
  298. parameters = oozie_workflow.conf_dict.copy()
  299. for action in oozie_workflow.actions:
  300. action.oozie_coordinator = oozie_coordinator
  301. action.oozie_bundle = oozie_bundle
  302. if request.GET.get('format') == 'json':
  303. if not workflow_graph and request.GET.get('is_jb2'):
  304. workflow_graph = django_mako.render_to_string('dashboard/list_oozie_workflow_graph.mako', {})
  305. return_obj = {
  306. 'id': oozie_workflow.id,
  307. 'status': oozie_workflow.status,
  308. 'progress': oozie_workflow.get_progress(full_node_list),
  309. 'graph': workflow_graph,
  310. 'actions': massaged_workflow_actions_for_json(oozie_workflow.get_working_actions(), oozie_coordinator, oozie_bundle),
  311. 'doc_url': doc.get_absolute_url() if doc else '',
  312. }
  313. return JsonResponse(return_obj, encoder=JSONEncoderForHTML)
  314. if request.GET.get('format') == 'svg':
  315. oozie_api = get_oozie(request.user, api_version="v2")
  316. svg_data = oozie_api.get_job_graph(job_id)
  317. return HttpResponse(svg_data)
  318. if request.GET.get('graph'):
  319. return render('dashboard/list_oozie_workflow_graph.mako', request, {
  320. 'oozie_workflow': oozie_workflow,
  321. 'workflow_graph': workflow_graph,
  322. 'layout_json': json.dumps(workflow_data.get('layout', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  323. 'workflow_json': json.dumps(workflow_data.get('workflow', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  324. 'credentials_json': json.dumps(credentials.credentials.keys(), cls=JSONEncoderForHTML) if credentials else '',
  325. 'workflow_properties_json': json.dumps(WORKFLOW_NODE_PROPERTIES, cls=JSONEncoderForHTML),
  326. 'doc_uuid': doc.uuid if doc else '',
  327. 'graph_element_id': request.GET.get('element') if request.GET.get('element') else 'loaded ' + doc.uuid + ' graph',
  328. 'subworkflows_json': json.dumps(_get_workflows(request.user), cls=JSONEncoderForHTML),
  329. 'can_edit_json': json.dumps(doc is None or doc.doc.get().is_editable(request.user)),
  330. 'is_jb2': request.GET.get('is_jb2', False)
  331. })
  332. oozie_slas = []
  333. if oozie_workflow.has_sla:
  334. oozie_api = get_oozie(request.user, api_version="v2")
  335. params = {
  336. 'id': oozie_workflow.id,
  337. 'parent_id': oozie_workflow.id
  338. }
  339. oozie_slas = oozie_api.get_oozie_slas(**params)
  340. return render('dashboard/list_oozie_workflow.mako', request, {
  341. 'oozie_workflow': oozie_workflow,
  342. 'oozie_coordinator': oozie_coordinator,
  343. 'oozie_bundle': oozie_bundle,
  344. 'oozie_parent': oozie_parent,
  345. 'oozie_slas': oozie_slas,
  346. 'hue_workflow': hue_workflow,
  347. 'hue_coord': hue_coord,
  348. 'parameters': dict((var, val) for var, val in parameters.iteritems() if var not in ParameterForm.NON_PARAMETERS and var != 'oozie.use.system.libpath' or var == 'oozie.wf.application.path'),
  349. 'has_job_edition_permission': has_job_edition_permission,
  350. 'workflow_graph': workflow_graph,
  351. 'layout_json': json.dumps(workflow_data.get('layout', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  352. 'workflow_json': json.dumps(workflow_data.get('workflow', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  353. 'credentials_json': json.dumps(credentials.credentials.keys(), cls=JSONEncoderForHTML) if credentials else '',
  354. 'workflow_properties_json': json.dumps(WORKFLOW_NODE_PROPERTIES, cls=JSONEncoderForHTML),
  355. 'doc_uuid': doc.uuid if doc else '',
  356. 'subworkflows_json': json.dumps(_get_workflows(request.user), cls=JSONEncoderForHTML),
  357. 'can_edit_json': json.dumps(doc is None or doc.doc.get().is_editable(request.user))
  358. })
  359. @show_oozie_error
  360. def list_oozie_coordinator(request, job_id):
  361. kwargs = {'cnt': 50, 'filters': []}
  362. kwargs['offset'] = request.GET.get('offset', 1)
  363. if request.GET.getlist('status'):
  364. kwargs['filters'].extend([('status', status) for status in request.GET.getlist('status')])
  365. oozie_coordinator = check_job_access_permission(request, job_id, **kwargs)
  366. # Cross reference the submission history (if any)
  367. coordinator = get_history().get_coordinator_from_config(oozie_coordinator.conf_dict)
  368. try:
  369. if not ENABLE_V2.get():
  370. coordinator = get_history().objects.get(oozie_job_id=job_id).job.get_full_node()
  371. except:
  372. LOG.exception("Ignoring error getting oozie job coordinator for job_id=%s", job_id)
  373. oozie_bundle = None
  374. if request.GET.get('bundle_job_id'):
  375. try:
  376. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  377. except:
  378. LOG.exception("Ignoring error getting oozie bundle for job_id=%s", job_id)
  379. if request.GET.get('format') == 'json':
  380. actions = massaged_coordinator_actions_for_json(oozie_coordinator, oozie_bundle)
  381. return_obj = {
  382. 'id': oozie_coordinator.id,
  383. 'status': oozie_coordinator.status,
  384. 'progress': oozie_coordinator.get_progress(),
  385. 'nextTime': format_time(oozie_coordinator.nextMaterializedTime),
  386. 'endTime': format_time(oozie_coordinator.endTime),
  387. 'actions': actions,
  388. 'total_actions': oozie_coordinator.total,
  389. 'doc_url': coordinator.get_absolute_url() if coordinator else '',
  390. }
  391. return JsonResponse(return_obj, encoder=JSONEncoderForHTML)
  392. oozie_slas = []
  393. if oozie_coordinator.has_sla:
  394. oozie_api = get_oozie(request.user, api_version="v2")
  395. params = {
  396. 'id': oozie_coordinator.id,
  397. 'parent_id': oozie_coordinator.id
  398. }
  399. oozie_slas = oozie_api.get_oozie_slas(**params)
  400. enable_cron_scheduling = ENABLE_CRON_SCHEDULING.get()
  401. update_coord_form = UpdateCoordinatorForm(oozie_coordinator=oozie_coordinator)
  402. return render('dashboard/list_oozie_coordinator.mako', request, {
  403. 'oozie_coordinator': oozie_coordinator,
  404. 'oozie_slas': oozie_slas,
  405. 'coordinator': coordinator,
  406. 'oozie_bundle': oozie_bundle,
  407. 'has_job_edition_permission': has_job_edition_permission,
  408. 'enable_cron_scheduling': enable_cron_scheduling,
  409. 'update_coord_form': update_coord_form,
  410. })
  411. @show_oozie_error
  412. def list_oozie_bundle(request, job_id):
  413. oozie_bundle = check_job_access_permission(request, job_id)
  414. # Cross reference the submission history (if any)
  415. bundle = None
  416. try:
  417. if ENABLE_V2.get():
  418. bundle = get_history().get_bundle_from_config(oozie_bundle.conf_dict)
  419. else:
  420. bundle = get_history().objects.get(oozie_job_id=job_id).job.get_full_node()
  421. except:
  422. LOG.exception("Ignoring error getting oozie job bundle for job_id=%s", job_id)
  423. if request.GET.get('format') == 'json':
  424. return_obj = {
  425. 'id': oozie_bundle.id,
  426. 'user': oozie_bundle.user,
  427. 'name': oozie_bundle.bundleJobName,
  428. 'status': oozie_bundle.status,
  429. 'progress': oozie_bundle.get_progress(),
  430. 'endTime': format_time(oozie_bundle.endTime),
  431. 'actions': massaged_bundle_actions_for_json(oozie_bundle),
  432. 'submitted': format_time(oozie_bundle.kickoffTime),
  433. 'doc_url': bundle.get_absolute_url() if bundle else '',
  434. 'canEdit': has_job_edition_permission(oozie_bundle, request.user),
  435. }
  436. return HttpResponse(json.dumps(return_obj).replace('\\\\', '\\'), content_type="application/json")
  437. return render('dashboard/list_oozie_bundle.mako', request, {
  438. 'oozie_bundle': oozie_bundle,
  439. 'bundle': bundle,
  440. 'has_job_edition_permission': has_job_edition_permission,
  441. })
  442. @show_oozie_error
  443. def list_oozie_workflow_action(request, action):
  444. try:
  445. action = get_oozie(request.user).get_action(action)
  446. workflow = check_job_access_permission(request, action.id.split('@')[0])
  447. except RestException, ex:
  448. msg = _("Error accessing Oozie action %s.") % (action,)
  449. LOG.exception(msg)
  450. raise PopupException(msg, detail=ex.message)
  451. oozie_coordinator = None
  452. if request.GET.get('coordinator_job_id'):
  453. oozie_coordinator = check_job_access_permission(request, request.GET.get('coordinator_job_id'))
  454. oozie_bundle = None
  455. if request.GET.get('bundle_job_id'):
  456. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  457. workflow.oozie_coordinator = oozie_coordinator
  458. workflow.oozie_bundle = oozie_bundle
  459. oozie_parent = workflow.get_parent_job_id()
  460. if oozie_parent:
  461. oozie_parent = check_job_access_permission(request, oozie_parent)
  462. return render('dashboard/list_oozie_workflow_action.mako', request, {
  463. 'action': action,
  464. 'workflow': workflow,
  465. 'oozie_coordinator': oozie_coordinator,
  466. 'oozie_bundle': oozie_bundle,
  467. 'oozie_parent': oozie_parent,
  468. })
  469. @show_oozie_error
  470. def get_oozie_job_log(request, job_id):
  471. oozie_api = get_oozie(request.user, api_version="v2")
  472. check_job_access_permission(request, job_id)
  473. kwargs = {'logfilter' : []}
  474. if request.GET.get('format') == 'json':
  475. if request.GET.get('recent'):
  476. kwargs['logfilter'].extend([('recent', val) for val in request.GET.get('recent').split(':')])
  477. if request.GET.get('limit'):
  478. kwargs['logfilter'].extend([('limit', request.GET.get('limit'))])
  479. if request.GET.get('loglevel'):
  480. kwargs['logfilter'].extend([('loglevel', request.GET.get('loglevel'))])
  481. if request.GET.get('text'):
  482. kwargs['logfilter'].extend([('text', request.GET.get('text'))])
  483. status_resp = oozie_api.get_job_status(job_id)
  484. log = oozie_api.get_job_log(job_id, **kwargs)
  485. return_obj = {
  486. 'id': job_id,
  487. 'status': status_resp['status'],
  488. 'log': log,
  489. }
  490. return JsonResponse(return_obj, encoder=JSONEncoderForHTML)
  491. @show_oozie_error
  492. def list_oozie_info(request):
  493. api = get_oozie(request.user)
  494. configuration = api.get_configuration()
  495. oozie_status = api.get_oozie_status()
  496. instrumentation = {}
  497. metrics = {}
  498. if 'org.apache.oozie.service.MetricsInstrumentationService' in [c.strip() for c in configuration.get('oozie.services.ext', '').split(',')]:
  499. api2 = get_oozie(request.user, api_version="v2")
  500. metrics = api2.get_metrics()
  501. else:
  502. instrumentation = api.get_instrumentation()
  503. return render('dashboard/list_oozie_info.mako', request, {
  504. 'instrumentation': instrumentation,
  505. 'metrics': metrics,
  506. 'configuration': configuration,
  507. 'oozie_status': oozie_status,
  508. 'is_embeddable': request.GET.get('is_embeddable', False),
  509. })
  510. @show_oozie_error
  511. def list_oozie_sla(request):
  512. oozie_api = get_oozie(request.user, api_version="v2")
  513. if request.method == 'POST':
  514. params = {}
  515. job_name = request.POST.get('job_name')
  516. if re.match('.*-oozie-\w+-[WCB]', job_name):
  517. params['id'] = job_name
  518. params['parent_id'] = job_name
  519. else:
  520. params['app_name'] = job_name
  521. if 'useDates' in request.POST:
  522. if request.POST.get('start'):
  523. params['nominal_start'] = request.POST.get('start')
  524. if request.POST.get('end'):
  525. params['nominal_end'] = request.POST.get('end')
  526. oozie_slas = oozie_api.get_oozie_slas(**params)
  527. else:
  528. oozie_slas = [] # or get latest?
  529. if request.GET.get('format') == 'json':
  530. massaged_slas = []
  531. for sla in oozie_slas:
  532. massaged_slas.append(massaged_sla_for_json(sla))
  533. return HttpResponse(json.dumps({'oozie_slas': massaged_slas}), content_type="text/json")
  534. configuration = oozie_api.get_configuration()
  535. show_slas_hint = 'org.apache.oozie.sla.service.SLAService' not in configuration.get('oozie.services.ext', '')
  536. return render('dashboard/list_oozie_sla.mako', request, {
  537. 'oozie_slas': oozie_slas,
  538. 'show_slas_hint': show_slas_hint,
  539. 'is_embeddable': request.GET.get('is_embeddable', False),
  540. })
  541. def massaged_sla_for_json(sla):
  542. massaged_sla = {
  543. 'slaStatus': sla['slaStatus'],
  544. 'id': sla['id'],
  545. 'appType': sla['appType'],
  546. 'appName': sla['appName'],
  547. 'appUrl': get_link(sla['id']),
  548. 'user': sla['user'],
  549. 'nominalTime': sla['nominalTime'],
  550. 'expectedStart': sla['expectedStart'],
  551. 'actualStart': sla['actualStart'],
  552. 'expectedEnd': sla['expectedEnd'],
  553. 'actualEnd': sla['actualEnd'],
  554. 'jobStatus': sla['jobStatus'],
  555. 'expectedDuration': sla['expectedDuration'],
  556. 'actualDuration': sla['actualDuration'],
  557. 'lastModified': sla['lastModified']
  558. }
  559. return massaged_sla
  560. @show_oozie_error
  561. def sync_coord_workflow(request, job_id):
  562. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  563. job = check_job_access_permission(request, job_id)
  564. check_job_edition_permission(job, request.user)
  565. hue_coord = get_history().get_coordinator_from_config(job.conf_dict)
  566. hue_wf = (hue_coord and hue_coord.workflow) or get_history().get_workflow_from_config(job.conf_dict)
  567. wf_application_path = job.conf_dict.get('wf_application_path') and Hdfs.urlsplit(job.conf_dict['wf_application_path'])[2] or ''
  568. coord_application_path = job.conf_dict.get('oozie.coord.application.path') and Hdfs.urlsplit(job.conf_dict['oozie.coord.application.path'])[2] or ''
  569. properties = hue_coord and hue_coord.properties and dict([(param['name'], param['value']) for param in hue_coord.properties]) or None
  570. if request.method == 'POST':
  571. params_form = ParametersFormSet(request.POST)
  572. if params_form.is_valid():
  573. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  574. # Update workflow params in coordinator
  575. hue_coord.clear_workflow_params()
  576. properties = dict([(param['name'], param['value']) for param in hue_coord.properties])
  577. # Deploy WF XML
  578. submission = Submission(user=request.user, job=hue_wf, fs=request.fs, jt=request.jt, properties=properties)
  579. submission.deploy(deployment_dir=wf_application_path)
  580. submission._create_file(wf_application_path, hue_wf.XML_FILE_NAME, hue_wf.to_xml(mapping=properties), do_as=True)
  581. # Deploy Coordinator XML
  582. job.conf_dict.update(mapping)
  583. submission = Submission(user=request.user, job=hue_coord, fs=request.fs, jt=request.jt, properties=job.conf_dict, oozie_id=job.id)
  584. submission._create_file(coord_application_path, hue_coord.XML_FILE_NAME, hue_coord.to_xml(mapping=job.conf_dict), do_as=True)
  585. # Server picks up deployed Coordinator XML changes after running 'update' action
  586. submission.update_coord()
  587. request.info(_('Successfully updated Workflow definition'))
  588. return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
  589. else:
  590. request.error(_('Invalid submission form: %s' % params_form.errors))
  591. else:
  592. new_params = hue_wf and hue_wf.find_all_parameters() or []
  593. new_params = dict([(param['name'], param['value']) for param in new_params])
  594. # Set previous values
  595. if properties:
  596. new_params = dict([(key, properties[key]) if key in properties.keys() else (key, new_params[key]) for key, value in new_params.iteritems()])
  597. initial_params = ParameterForm.get_initial_params(new_params)
  598. params_form = ParametersFormSet(initial=initial_params)
  599. popup = render('editor2/submit_job_popup.mako', request, {
  600. 'params_form': params_form,
  601. 'name': _('Job'),
  602. 'header': _('Sync Workflow definition?'),
  603. 'action': reverse('oozie:sync_coord_workflow', kwargs={'job_id': job_id})
  604. }, force_template=True).content
  605. return JsonResponse(popup, safe=False)
  606. @show_oozie_error
  607. def rerun_oozie_job(request, job_id, app_path=None):
  608. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  609. oozie_workflow = check_job_access_permission(request, job_id)
  610. check_job_edition_permission(oozie_workflow, request.user)
  611. if app_path is None:
  612. app_path = oozie_workflow.appPath
  613. return_json = request.GET.get('format') == 'json'
  614. if request.method == 'POST':
  615. rerun_form = RerunForm(request.POST, oozie_workflow=oozie_workflow)
  616. params_form = ParametersFormSet(request.POST)
  617. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  618. args = {}
  619. if request.POST.get('rerun_form_choice') == 'fail_nodes':
  620. args['fail_nodes'] = 'true'
  621. else:
  622. args['skip_nodes'] = ','.join(rerun_form.cleaned_data['skip_nodes'])
  623. args['deployment_dir'] = app_path
  624. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  625. _rerun_workflow(request, job_id, args, mapping)
  626. if rerun_form.cleaned_data['return_json']:
  627. return JsonResponse({'status': 0, 'job_id': job_id}, safe=False)
  628. else:
  629. request.info(_('Workflow re-running.'))
  630. return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_id}))
  631. else:
  632. request.error(_('Invalid submission form: %s %s' % (rerun_form.errors, params_form.errors)))
  633. else:
  634. rerun_form = RerunForm(oozie_workflow=oozie_workflow, return_json=return_json)
  635. initial_params = ParameterForm.get_initial_params(oozie_workflow.conf_dict)
  636. params_form = ParametersFormSet(initial=initial_params)
  637. popup = render('dashboard/rerun_workflow_popup.mako', request, {
  638. 'rerun_form': rerun_form,
  639. 'params_form': params_form,
  640. 'action': reverse('oozie:rerun_oozie_job', kwargs={'job_id': job_id, 'app_path': app_path}),
  641. 'return_json': return_json
  642. }, force_template=True).content
  643. return JsonResponse(popup, safe=False)
  644. def _rerun_workflow(request, oozie_id, run_args, mapping):
  645. try:
  646. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, properties=mapping, oozie_id=oozie_id)
  647. job_id = submission.rerun(**run_args)
  648. return job_id
  649. except RestException, ex:
  650. msg = _("Error re-running workflow %s.") % (oozie_id,)
  651. LOG.exception(msg)
  652. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  653. @show_oozie_error
  654. def rerun_oozie_coordinator(request, job_id, app_path=None):
  655. oozie_coordinator = check_job_access_permission(request, job_id)
  656. check_job_edition_permission(oozie_coordinator, request.user)
  657. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  658. if app_path is None:
  659. app_path = oozie_coordinator.coordJobPath
  660. return_json = request.GET.get('format') == 'json'
  661. if request.method == 'POST':
  662. params_form = ParametersFormSet(request.POST)
  663. rerun_form = RerunCoordForm(request.POST, oozie_coordinator=oozie_coordinator)
  664. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  665. args = {}
  666. args['deployment_dir'] = app_path
  667. params = {
  668. 'type': 'action',
  669. 'scope': ','.join(oozie_coordinator.aggreate(rerun_form.cleaned_data['actions'])),
  670. 'refresh': rerun_form.cleaned_data['refresh'],
  671. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  672. }
  673. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  674. _rerun_coordinator(request, job_id, args, params, properties)
  675. if rerun_form.cleaned_data['return_json']:
  676. return JsonResponse({'status': 0, 'job_id': job_id}, safe=False)
  677. else:
  678. request.info(_('Coordinator re-running.'))
  679. return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
  680. else:
  681. request.error(_('Invalid submission form: %s') % smart_unicode(rerun_form.errors))
  682. return list_oozie_coordinator(request, job_id)
  683. else:
  684. rerun_form = RerunCoordForm(oozie_coordinator=oozie_coordinator, return_json=return_json)
  685. initial_params = ParameterForm.get_initial_params(oozie_coordinator.conf_dict)
  686. params_form = ParametersFormSet(initial=initial_params)
  687. popup = render('dashboard/rerun_coord_popup.mako', request, {
  688. 'rerun_form': rerun_form,
  689. 'params_form': params_form,
  690. 'action': reverse('oozie:rerun_oozie_coord', kwargs={'job_id': job_id, 'app_path': app_path}),
  691. 'return_json': return_json,
  692. }, force_template=True).content
  693. return JsonResponse(popup, safe=False)
  694. def _rerun_coordinator(request, oozie_id, args, params, properties):
  695. try:
  696. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  697. job_id = submission.rerun_coord(params=params, **args)
  698. return job_id
  699. except RestException, ex:
  700. msg = _("Error re-running coordinator %s.") % (oozie_id,)
  701. LOG.exception(msg)
  702. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  703. @show_oozie_error
  704. def rerun_oozie_bundle(request, job_id, app_path):
  705. oozie_bundle = check_job_access_permission(request, job_id)
  706. check_job_edition_permission(oozie_bundle, request.user)
  707. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  708. if request.method == 'POST':
  709. params_form = ParametersFormSet(request.POST)
  710. rerun_form = RerunBundleForm(request.POST, oozie_bundle=oozie_bundle)
  711. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  712. args = {}
  713. args['deployment_dir'] = app_path
  714. params = {
  715. 'coord-scope': ','.join(rerun_form.cleaned_data['coordinators']),
  716. 'refresh': rerun_form.cleaned_data['refresh'],
  717. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  718. }
  719. if rerun_form.cleaned_data['start'] and rerun_form.cleaned_data['end']:
  720. date = {
  721. 'date-scope':
  722. '%(start)s::%(end)s' % {
  723. 'start': utc_datetime_format(rerun_form.cleaned_data['start']),
  724. 'end': utc_datetime_format(rerun_form.cleaned_data['end'])
  725. }
  726. }
  727. params.update(date)
  728. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  729. _rerun_bundle(request, job_id, args, params, properties)
  730. request.info(_('Bundle re-running.'))
  731. return redirect(reverse('oozie:list_oozie_bundle', kwargs={'job_id': job_id}))
  732. else:
  733. request.error(_('Invalid submission form: %s' % (rerun_form.errors,)))
  734. return list_oozie_bundle(request, job_id)
  735. else:
  736. rerun_form = RerunBundleForm(oozie_bundle=oozie_bundle)
  737. initial_params = ParameterForm.get_initial_params(oozie_bundle.conf_dict)
  738. params_form = ParametersFormSet(initial=initial_params)
  739. popup = render('dashboard/rerun_bundle_popup.mako', request, {
  740. 'rerun_form': rerun_form,
  741. 'params_form': params_form,
  742. 'action': reverse('oozie:rerun_oozie_bundle', kwargs={'job_id': job_id, 'app_path': app_path}),
  743. }, force_template=True).content
  744. return JsonResponse(popup, safe=False)
  745. def _rerun_bundle(request, oozie_id, args, params, properties):
  746. try:
  747. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  748. job_id = submission.rerun_bundle(params=params, **args)
  749. return job_id
  750. except RestException, ex:
  751. msg = _("Error re-running bundle %s.") % (oozie_id,)
  752. LOG.exception(msg)
  753. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  754. def submit_external_job(request, application_path):
  755. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  756. if request.method == 'POST':
  757. params_form = ParametersFormSet(request.POST)
  758. if params_form.is_valid():
  759. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  760. mapping['dryrun'] = request.POST.get('dryrun_checkbox') == 'on'
  761. application_name = os.path.basename(application_path)
  762. application_class = Bundle if application_name == 'bundle.xml' else Coordinator if application_name == 'coordinator.xml' else get_workflow()
  763. mapping[application_class.get_application_path_key()] = os.path.dirname(application_path)
  764. try:
  765. submission = Submission(request.user, fs=request.fs, jt=request.jt, properties=mapping)
  766. job_id = submission.run(application_path)
  767. except RestException, ex:
  768. detail = ex._headers.get('oozie-error-message', ex)
  769. if 'Max retries exceeded with url' in str(detail):
  770. detail = '%s: %s' % (_('The Oozie server is not running'), detail)
  771. LOG.exception(smart_str(detail))
  772. raise PopupException(_("Error submitting job %s") % (application_path,), detail=detail)
  773. jsonify = request.POST.get('format') == 'json'
  774. if jsonify:
  775. return JsonResponse({'status': 0, 'job_id': job_id, 'type': 'external_workflow'}, safe=False)
  776. else:
  777. request.info(_('Oozie job submitted'))
  778. view = 'list_oozie_bundle' if application_name == 'bundle.xml' else 'list_oozie_coordinator' if application_name == 'coordinator.xml' else 'list_oozie_workflow'
  779. return redirect(reverse('oozie:%s' % view, kwargs={'job_id': job_id}))
  780. else:
  781. request.error(_('Invalid submission form: %s' % params_form.errors))
  782. else:
  783. parameters = Submission(request.user, fs=request.fs, jt=request.jt).get_external_parameters(application_path)
  784. initial_params = ParameterForm.get_initial_params(parameters)
  785. params_form = ParametersFormSet(initial=initial_params)
  786. popup = render('editor/submit_job_popup.mako', request, {
  787. 'params_form': params_form,
  788. 'name': _('Job'),
  789. 'action': reverse('oozie:submit_external_job', kwargs={'application_path': application_path}),
  790. 'show_dryrun': os.path.basename(application_path) != 'bundle.xml',
  791. 'return_json': request.GET.get('format') == 'json'
  792. }, force_template=True).content
  793. return JsonResponse(popup, safe=False)
  794. def massaged_workflow_actions_for_json(workflow_actions, oozie_coordinator, oozie_bundle):
  795. actions = []
  796. for action in workflow_actions:
  797. if oozie_coordinator is not None:
  798. setattr(action, 'oozie_coordinator', oozie_coordinator)
  799. if oozie_bundle is not None:
  800. setattr(action, 'oozie_bundle', oozie_bundle)
  801. massaged_action = {
  802. 'id': action.id,
  803. 'log': action.get_absolute_log_url(),
  804. 'url': action.get_absolute_url(),
  805. 'name': action.name,
  806. 'type': action.type,
  807. 'status': action.status,
  808. 'externalIdUrl': action.get_external_id_url(),
  809. 'externalId': action.externalId,
  810. 'externalJobId': action.externalId,
  811. 'startTime': format_time(action.startTime),
  812. 'endTime': format_time(action.endTime),
  813. 'retries': action.retries,
  814. 'errorCode': action.errorCode,
  815. 'errorMessage': action.errorMessage,
  816. 'transition': action.transition,
  817. 'data': action.data,
  818. }
  819. actions.append(massaged_action)
  820. return actions
  821. def massaged_coordinator_actions_for_json(coordinator, oozie_bundle):
  822. coordinator_id = coordinator.id
  823. coordinator_actions = coordinator.get_working_actions()
  824. actions = []
  825. related_job_ids = []
  826. related_job_ids.append('coordinator_job_id=%s' % coordinator_id)
  827. if oozie_bundle is not None:
  828. related_job_ids.append('bundle_job_id=%s' %oozie_bundle.id)
  829. for action in coordinator_actions:
  830. massaged_action = {
  831. 'id': action.id,
  832. 'url': action.externalId and reverse('oozie:list_oozie_workflow', kwargs={'job_id': action.externalId}) + '?%s' % '&'.join(related_job_ids) or '',
  833. 'number': action.actionNumber,
  834. 'type': 'schedule-task',
  835. 'status': action.status,
  836. 'externalId': action.externalId or '-',
  837. 'externalIdUrl': action.externalId and reverse('oozie:list_oozie_workflow_action', kwargs={'action': action.externalId}) or '',
  838. 'nominalTime': format_time(action.nominalTime),
  839. 'title': action.title,
  840. 'createdTime': format_time(action.createdTime),
  841. 'lastModifiedTime': format_time(action.lastModifiedTime),
  842. 'errorCode': action.errorCode,
  843. 'errorMessage': action.errorMessage,
  844. 'missingDependencies': action.missingDependencies
  845. }
  846. actions.append(massaged_action)
  847. # Sorting for Oozie < 4.1 backward compatibility
  848. actions.sort(key=lambda k: k['number'], reverse=True)
  849. return actions
  850. def massaged_bundle_actions_for_json(bundle):
  851. bundle_actions = bundle.get_working_actions()
  852. actions = []
  853. for action in bundle_actions:
  854. massaged_action = {
  855. 'id': action.coordJobId,
  856. 'url': action.coordJobId and reverse('oozie:list_oozie_coordinator', kwargs={'job_id': action.coordJobId}) + '?bundle_job_id=%s' % bundle.id or '',
  857. 'name': action.coordJobName,
  858. 'type': action.type,
  859. 'status': action.status,
  860. 'externalId': action.coordExternalId or '-',
  861. 'frequency': action.frequency,
  862. 'timeUnit': action.timeUnit,
  863. 'nextMaterializedTime': action.nextMaterializedTime,
  864. 'concurrency': action.concurrency,
  865. 'pauseTime': action.pauseTime,
  866. 'user': action.user,
  867. 'acl': action.acl,
  868. 'timeOut': action.timeOut,
  869. 'coordJobPath': action.coordJobPath,
  870. 'executionPolicy': action.executionPolicy,
  871. 'startTime': action.startTime,
  872. 'endTime': action.endTime,
  873. 'lastAction': action.lastAction
  874. }
  875. actions.insert(0, massaged_action)
  876. return actions
  877. def format_time(st_time):
  878. if st_time is None:
  879. return '-'
  880. elif type(st_time) == time.struct_time:
  881. return time.strftime("%a, %d %b %Y %H:%M:%S", st_time)
  882. else:
  883. return st_time
  884. def massaged_oozie_jobs_for_json(oozie_jobs, user, just_sla=False):
  885. jobs = []
  886. for job in oozie_jobs:
  887. if not just_sla or (just_sla and job.has_sla) and job.appName != 'pig-app-hue-script':
  888. last_modified_time_millis = hasattr(job, 'lastModTime') and job.lastModTime and (time.time() - time.mktime(job.lastModTime)) * 1000 or 0
  889. duration_millis = job.durationTime
  890. massaged_job = {
  891. 'id': job.id,
  892. 'lastModTime': hasattr(job, 'lastModTime') and job.lastModTime and format_time(job.lastModTime) or None,
  893. 'lastModTimeInMillis': last_modified_time_millis,
  894. 'lastModTimeFormatted': last_modified_time_millis and format_duration_in_millis(last_modified_time_millis) or None,
  895. 'kickoffTime': hasattr(job, 'kickoffTime') and job.kickoffTime and format_time(job.kickoffTime) or '',
  896. 'kickoffTimeInMillis': hasattr(job, 'kickoffTime') and job.kickoffTime and time.mktime(catch_unicode_time(job.kickoffTime)) or 0,
  897. 'nextMaterializedTime': hasattr(job, 'nextMaterializedTime') and job.nextMaterializedTime and format_time(job.nextMaterializedTime) or '',
  898. 'nextMaterializedTimeInMillis': hasattr(job, 'nextMaterializedTime') and job.nextMaterializedTime and time.mktime(job.nextMaterializedTime) or 0,
  899. 'timeOut': hasattr(job, 'timeOut') and job.timeOut or None,
  900. 'endTime': job.endTime and format_time(job.endTime) or None,
  901. 'pauseTime': hasattr(job, 'pauseTime') and job.pauseTime and format_time(job.endTime) or None,
  902. 'concurrency': hasattr(job, 'concurrency') and job.concurrency or None,
  903. 'endTimeInMillis': job.endTime and time.mktime(job.endTime) or 0,
  904. 'lastActionInMillis': hasattr(job, 'lastAction') and job.lastAction and time.mktime(job.lastAction) or 0,
  905. 'status': job.status,
  906. 'group': job.group,
  907. 'isRunning': job.is_running(),
  908. 'duration': duration_millis and format_duration_in_millis(duration_millis) or None,
  909. 'durationInMillis': duration_millis,
  910. 'appName': job.appName,
  911. 'progress': job.get_progress(),
  912. 'user': job.user,
  913. 'absoluteUrl': job.get_absolute_url(),
  914. 'canEdit': has_job_edition_permission(job, user),
  915. 'killUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'kill'}),
  916. 'suspendUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'suspend'}),
  917. 'resumeUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'resume'}),
  918. 'created': hasattr(job, 'createdTime') and job.createdTime and format_time(job.createdTime) or '',
  919. 'createdInMillis': job.submissionTime,
  920. 'startTime': hasattr(job, 'startTime') and format_time(job.startTime) or None,
  921. 'startTimeInMillis': hasattr(job, 'startTime') and job.startTime and time.mktime(job.startTime) or 0,
  922. 'run': hasattr(job, 'run') and job.run or 0,
  923. 'frequency': hasattr(job, 'frequency') and Coordinator.CRON_MAPPING.get(job.frequency, job.frequency) or None,
  924. 'timeUnit': hasattr(job, 'timeUnit') and job.timeUnit or None,
  925. 'parentUrl': hasattr(job, 'parentId') and job.parentId and get_link(job.parentId) or '',
  926. 'submittedManually': hasattr(job, 'parentId') and (job.parentId is None or 'C@' not in job.parentId)
  927. }
  928. jobs.append(massaged_job)
  929. return { 'jobs': jobs }
  930. def check_job_access_permission(request, job_id, **kwargs):
  931. """
  932. Decorator ensuring that the user has access to the job submitted to Oozie.
  933. Arg: Oozie 'workflow', 'coordinator' or 'bundle' ID.
  934. Return: the Oozie workflow, coordinator or bundle or raise an exception
  935. Notice: its gets an id in input and returns the full object in output (not an id).
  936. """
  937. if job_id is not None:
  938. oozie_api = get_oozie(request.user)
  939. if job_id.endswith('W'):
  940. get_job = oozie_api.get_job
  941. elif job_id.endswith('C'):
  942. get_job = oozie_api.get_coordinator
  943. else:
  944. get_job = oozie_api.get_bundle
  945. try:
  946. if job_id.endswith('C'):
  947. oozie_job = get_job(job_id, **kwargs)
  948. else:
  949. oozie_job = get_job(job_id)
  950. except RestException, ex:
  951. msg = _("Error accessing Oozie job %s.") % (job_id,)
  952. LOG.exception(msg)
  953. raise PopupException(msg, detail=ex._headers.get('oozie-error-message'))
  954. if request.user.is_superuser \
  955. or oozie_job.user == request.user.username \
  956. or has_dashboard_jobs_access(request.user):
  957. return oozie_job
  958. else:
  959. message = _("Permission denied. %(username)s does not have the permissions to access job %(id)s.") % \
  960. {'username': request.user.username, 'id': oozie_job.id}
  961. access_warn(request, message)
  962. raise PopupException(message)
  963. def check_job_edition_permission(oozie_job, user):
  964. if has_job_edition_permission(oozie_job, user):
  965. return oozie_job
  966. else:
  967. message = _("Permission denied. %(username)s does not have the permissions to modify job %(id)s.") % \
  968. {'username': user.username, 'id': oozie_job.id}
  969. raise PopupException(message)
  970. def has_job_edition_permission(oozie_job, user):
  971. return user.is_superuser or oozie_job.user == user.username or (oozie_job.group and user.groups.filter(name=oozie_job.group).exists()) or (oozie_job.acl and user.username in oozie_job.acl.split(','))
  972. def has_dashboard_jobs_access(user):
  973. return user.is_superuser or user.has_hue_permission(action="dashboard_jobs_access", app=DJANGO_APPS[0])