dashboard.py 47 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194
  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.core.urlresolvers 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('graph'):
  315. return render('dashboard/list_oozie_workflow_graph.mako', request, {
  316. 'oozie_workflow': oozie_workflow,
  317. 'workflow_graph': workflow_graph,
  318. 'layout_json': json.dumps(workflow_data.get('layout', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  319. 'workflow_json': json.dumps(workflow_data.get('workflow', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  320. 'credentials_json': json.dumps(credentials.credentials.keys(), cls=JSONEncoderForHTML) if credentials else '',
  321. 'workflow_properties_json': json.dumps(WORKFLOW_NODE_PROPERTIES, cls=JSONEncoderForHTML),
  322. 'doc_uuid': doc.uuid if doc else '',
  323. 'graph_element_id': request.GET.get('element') if request.GET.get('element') else 'loaded ' + doc.uuid + ' graph',
  324. 'subworkflows_json': json.dumps(_get_workflows(request.user), cls=JSONEncoderForHTML),
  325. 'can_edit_json': json.dumps(doc is None or doc.doc.get().is_editable(request.user)),
  326. 'is_jb2': request.GET.get('is_jb2', False)
  327. })
  328. oozie_slas = []
  329. if oozie_workflow.has_sla:
  330. oozie_api = get_oozie(request.user, api_version="v2")
  331. params = {
  332. 'id': oozie_workflow.id,
  333. 'parent_id': oozie_workflow.id
  334. }
  335. oozie_slas = oozie_api.get_oozie_slas(**params)
  336. return render('dashboard/list_oozie_workflow.mako', request, {
  337. 'oozie_workflow': oozie_workflow,
  338. 'oozie_coordinator': oozie_coordinator,
  339. 'oozie_bundle': oozie_bundle,
  340. 'oozie_parent': oozie_parent,
  341. 'oozie_slas': oozie_slas,
  342. 'hue_workflow': hue_workflow,
  343. 'hue_coord': hue_coord,
  344. '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'),
  345. 'has_job_edition_permission': has_job_edition_permission,
  346. 'workflow_graph': workflow_graph,
  347. 'layout_json': json.dumps(workflow_data.get('layout', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  348. 'workflow_json': json.dumps(workflow_data.get('workflow', ''), cls=JSONEncoderForHTML) if workflow_data else '',
  349. 'credentials_json': json.dumps(credentials.credentials.keys(), cls=JSONEncoderForHTML) if credentials else '',
  350. 'workflow_properties_json': json.dumps(WORKFLOW_NODE_PROPERTIES, cls=JSONEncoderForHTML),
  351. 'doc_uuid': doc.uuid if doc else '',
  352. 'subworkflows_json': json.dumps(_get_workflows(request.user), cls=JSONEncoderForHTML),
  353. 'can_edit_json': json.dumps(doc is None or doc.doc.get().is_editable(request.user))
  354. })
  355. @show_oozie_error
  356. def list_oozie_coordinator(request, job_id):
  357. kwargs = {'cnt': 50, 'filters': []}
  358. kwargs['offset'] = request.GET.get('offset', 1)
  359. if request.GET.getlist('status'):
  360. kwargs['filters'].extend([('status', status) for status in request.GET.getlist('status')])
  361. oozie_coordinator = check_job_access_permission(request, job_id, **kwargs)
  362. # Cross reference the submission history (if any)
  363. coordinator = get_history().get_coordinator_from_config(oozie_coordinator.conf_dict)
  364. try:
  365. if not ENABLE_V2.get():
  366. coordinator = get_history().objects.get(oozie_job_id=job_id).job.get_full_node()
  367. except:
  368. LOG.exception("Ignoring error getting oozie job coordinator for job_id=%s", job_id)
  369. oozie_bundle = None
  370. if request.GET.get('bundle_job_id'):
  371. try:
  372. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  373. except:
  374. LOG.exception("Ignoring error getting oozie bundle for job_id=%s", job_id)
  375. if request.GET.get('format') == 'json':
  376. actions = massaged_coordinator_actions_for_json(oozie_coordinator, oozie_bundle)
  377. return_obj = {
  378. 'id': oozie_coordinator.id,
  379. 'status': oozie_coordinator.status,
  380. 'progress': oozie_coordinator.get_progress(),
  381. 'nextTime': format_time(oozie_coordinator.nextMaterializedTime),
  382. 'endTime': format_time(oozie_coordinator.endTime),
  383. 'actions': actions,
  384. 'total_actions': oozie_coordinator.total,
  385. 'doc_url': coordinator.get_absolute_url() if coordinator else '',
  386. }
  387. return JsonResponse(return_obj, encoder=JSONEncoderForHTML)
  388. oozie_slas = []
  389. if oozie_coordinator.has_sla:
  390. oozie_api = get_oozie(request.user, api_version="v2")
  391. params = {
  392. 'id': oozie_coordinator.id,
  393. 'parent_id': oozie_coordinator.id
  394. }
  395. oozie_slas = oozie_api.get_oozie_slas(**params)
  396. enable_cron_scheduling = ENABLE_CRON_SCHEDULING.get()
  397. update_coord_form = UpdateCoordinatorForm(oozie_coordinator=oozie_coordinator)
  398. return render('dashboard/list_oozie_coordinator.mako', request, {
  399. 'oozie_coordinator': oozie_coordinator,
  400. 'oozie_slas': oozie_slas,
  401. 'coordinator': coordinator,
  402. 'oozie_bundle': oozie_bundle,
  403. 'has_job_edition_permission': has_job_edition_permission,
  404. 'enable_cron_scheduling': enable_cron_scheduling,
  405. 'update_coord_form': update_coord_form,
  406. })
  407. @show_oozie_error
  408. def list_oozie_bundle(request, job_id):
  409. oozie_bundle = check_job_access_permission(request, job_id)
  410. # Cross reference the submission history (if any)
  411. bundle = None
  412. try:
  413. if ENABLE_V2.get():
  414. bundle = get_history().get_bundle_from_config(oozie_bundle.conf_dict)
  415. else:
  416. bundle = get_history().objects.get(oozie_job_id=job_id).job.get_full_node()
  417. except:
  418. LOG.exception("Ignoring error getting oozie job bundle for job_id=%s", job_id)
  419. if request.GET.get('format') == 'json':
  420. return_obj = {
  421. 'id': oozie_bundle.id,
  422. 'user': oozie_bundle.user,
  423. 'name': oozie_bundle.bundleJobName,
  424. 'status': oozie_bundle.status,
  425. 'progress': oozie_bundle.get_progress(),
  426. 'endTime': format_time(oozie_bundle.endTime),
  427. 'actions': massaged_bundle_actions_for_json(oozie_bundle),
  428. 'submitted': format_time(oozie_bundle.kickoffTime),
  429. 'doc_url': bundle.get_absolute_url() if bundle else '',
  430. 'canEdit': has_job_edition_permission(oozie_bundle, request.user),
  431. }
  432. return HttpResponse(json.dumps(return_obj).replace('\\\\', '\\'), content_type="application/json")
  433. return render('dashboard/list_oozie_bundle.mako', request, {
  434. 'oozie_bundle': oozie_bundle,
  435. 'bundle': bundle,
  436. 'has_job_edition_permission': has_job_edition_permission,
  437. })
  438. @show_oozie_error
  439. def list_oozie_workflow_action(request, action):
  440. try:
  441. action = get_oozie(request.user).get_action(action)
  442. workflow = check_job_access_permission(request, action.id.split('@')[0])
  443. except RestException, ex:
  444. msg = _("Error accessing Oozie action %s.") % (action,)
  445. LOG.exception(msg)
  446. raise PopupException(msg, detail=ex.message)
  447. oozie_coordinator = None
  448. if request.GET.get('coordinator_job_id'):
  449. oozie_coordinator = check_job_access_permission(request, request.GET.get('coordinator_job_id'))
  450. oozie_bundle = None
  451. if request.GET.get('bundle_job_id'):
  452. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  453. workflow.oozie_coordinator = oozie_coordinator
  454. workflow.oozie_bundle = oozie_bundle
  455. oozie_parent = workflow.get_parent_job_id()
  456. if oozie_parent:
  457. oozie_parent = check_job_access_permission(request, oozie_parent)
  458. return render('dashboard/list_oozie_workflow_action.mako', request, {
  459. 'action': action,
  460. 'workflow': workflow,
  461. 'oozie_coordinator': oozie_coordinator,
  462. 'oozie_bundle': oozie_bundle,
  463. 'oozie_parent': oozie_parent,
  464. })
  465. @show_oozie_error
  466. def get_oozie_job_log(request, job_id):
  467. oozie_api = get_oozie(request.user, api_version="v2")
  468. check_job_access_permission(request, job_id)
  469. kwargs = {'logfilter' : []}
  470. if request.GET.get('format') == 'json':
  471. if request.GET.get('recent'):
  472. kwargs['logfilter'].extend([('recent', val) for val in request.GET.get('recent').split(':')])
  473. if request.GET.get('limit'):
  474. kwargs['logfilter'].extend([('limit', request.GET.get('limit'))])
  475. if request.GET.get('loglevel'):
  476. kwargs['logfilter'].extend([('loglevel', request.GET.get('loglevel'))])
  477. if request.GET.get('text'):
  478. kwargs['logfilter'].extend([('text', request.GET.get('text'))])
  479. status_resp = oozie_api.get_job_status(job_id)
  480. log = oozie_api.get_job_log(job_id, **kwargs)
  481. return_obj = {
  482. 'id': job_id,
  483. 'status': status_resp['status'],
  484. 'log': log,
  485. }
  486. return JsonResponse(return_obj, encoder=JSONEncoderForHTML)
  487. @show_oozie_error
  488. def list_oozie_info(request):
  489. api = get_oozie(request.user)
  490. configuration = api.get_configuration()
  491. oozie_status = api.get_oozie_status()
  492. instrumentation = {}
  493. metrics = {}
  494. if 'org.apache.oozie.service.MetricsInstrumentationService' in [c.strip() for c in configuration.get('oozie.services.ext', '').split(',')]:
  495. api2 = get_oozie(request.user, api_version="v2")
  496. metrics = api2.get_metrics()
  497. else:
  498. instrumentation = api.get_instrumentation()
  499. return render('dashboard/list_oozie_info.mako', request, {
  500. 'instrumentation': instrumentation,
  501. 'metrics': metrics,
  502. 'configuration': configuration,
  503. 'oozie_status': oozie_status,
  504. 'is_embeddable': request.GET.get('is_embeddable', False),
  505. })
  506. @show_oozie_error
  507. def list_oozie_sla(request):
  508. oozie_api = get_oozie(request.user, api_version="v2")
  509. if request.method == 'POST':
  510. params = {}
  511. job_name = request.POST.get('job_name')
  512. if re.match('.*-oozie-oozi-[WCB]', job_name):
  513. params['id'] = job_name
  514. params['parent_id'] = job_name
  515. else:
  516. params['app_name'] = job_name
  517. if 'useDates' in request.POST:
  518. if request.POST.get('start'):
  519. params['nominal_start'] = request.POST.get('start')
  520. if request.POST.get('end'):
  521. params['nominal_end'] = request.POST.get('end')
  522. oozie_slas = oozie_api.get_oozie_slas(**params)
  523. else:
  524. oozie_slas = [] # or get latest?
  525. if request.REQUEST.get('format') == 'json':
  526. massaged_slas = []
  527. for sla in oozie_slas:
  528. massaged_slas.append(massaged_sla_for_json(sla))
  529. return HttpResponse(json.dumps({'oozie_slas': massaged_slas}), content_type="text/json")
  530. configuration = oozie_api.get_configuration()
  531. show_slas_hint = 'org.apache.oozie.sla.service.SLAService' not in configuration.get('oozie.services.ext', '')
  532. return render('dashboard/list_oozie_sla.mako', request, {
  533. 'oozie_slas': oozie_slas,
  534. 'show_slas_hint': show_slas_hint,
  535. 'is_embeddable': request.GET.get('is_embeddable', False),
  536. })
  537. def massaged_sla_for_json(sla):
  538. massaged_sla = {
  539. 'slaStatus': sla['slaStatus'],
  540. 'id': sla['id'],
  541. 'appType': sla['appType'],
  542. 'appName': sla['appName'],
  543. 'appUrl': get_link(sla['id']),
  544. 'user': sla['user'],
  545. 'nominalTime': sla['nominalTime'],
  546. 'expectedStart': sla['expectedStart'],
  547. 'actualStart': sla['actualStart'],
  548. 'expectedEnd': sla['expectedEnd'],
  549. 'actualEnd': sla['actualEnd'],
  550. 'jobStatus': sla['jobStatus'],
  551. 'expectedDuration': sla['expectedDuration'],
  552. 'actualDuration': sla['actualDuration'],
  553. 'lastModified': sla['lastModified']
  554. }
  555. return massaged_sla
  556. @show_oozie_error
  557. def sync_coord_workflow(request, job_id):
  558. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  559. job = check_job_access_permission(request, job_id)
  560. check_job_edition_permission(job, request.user)
  561. hue_coord = get_history().get_coordinator_from_config(job.conf_dict)
  562. hue_wf = (hue_coord and hue_coord.workflow) or get_history().get_workflow_from_config(job.conf_dict)
  563. wf_application_path = job.conf_dict.get('wf_application_path') and Hdfs.urlsplit(job.conf_dict['wf_application_path'])[2] or ''
  564. coord_application_path = job.conf_dict.get('oozie.coord.application.path') and Hdfs.urlsplit(job.conf_dict['oozie.coord.application.path'])[2] or ''
  565. properties = hue_coord and hue_coord.properties and dict([(param['name'], param['value']) for param in hue_coord.properties]) or None
  566. if request.method == 'POST':
  567. params_form = ParametersFormSet(request.POST)
  568. if params_form.is_valid():
  569. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  570. # Update workflow params in coordinator
  571. hue_coord.clear_workflow_params()
  572. properties = dict([(param['name'], param['value']) for param in hue_coord.properties])
  573. # Deploy WF XML
  574. submission = Submission(user=request.user, job=hue_wf, fs=request.fs, jt=request.jt, properties=properties)
  575. submission.deploy(deployment_dir=wf_application_path)
  576. submission._create_file(wf_application_path, hue_wf.XML_FILE_NAME, hue_wf.to_xml(mapping=properties), do_as=True)
  577. # Deploy Coordinator XML
  578. job.conf_dict.update(mapping)
  579. submission = Submission(user=request.user, job=hue_coord, fs=request.fs, jt=request.jt, properties=job.conf_dict, oozie_id=job.id)
  580. submission._create_file(coord_application_path, hue_coord.XML_FILE_NAME, hue_coord.to_xml(mapping=job.conf_dict), do_as=True)
  581. # Server picks up deployed Coordinator XML changes after running 'update' action
  582. submission.update_coord()
  583. request.info(_('Successfully updated Workflow definition'))
  584. return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
  585. else:
  586. request.error(_('Invalid submission form: %s' % params_form.errors))
  587. else:
  588. new_params = hue_wf and hue_wf.find_all_parameters() or []
  589. new_params = dict([(param['name'], param['value']) for param in new_params])
  590. # Set previous values
  591. if properties:
  592. new_params = dict([(key, properties[key]) if key in properties.keys() else (key, new_params[key]) for key, value in new_params.iteritems()])
  593. initial_params = ParameterForm.get_initial_params(new_params)
  594. params_form = ParametersFormSet(initial=initial_params)
  595. popup = render('editor2/submit_job_popup.mako', request, {
  596. 'params_form': params_form,
  597. 'name': _('Job'),
  598. 'header': _('Sync Workflow definition?'),
  599. 'action': reverse('oozie:sync_coord_workflow', kwargs={'job_id': job_id})
  600. }, force_template=True).content
  601. return JsonResponse(popup, safe=False)
  602. @show_oozie_error
  603. def rerun_oozie_job(request, job_id, app_path=None):
  604. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  605. oozie_workflow = check_job_access_permission(request, job_id)
  606. check_job_edition_permission(oozie_workflow, request.user)
  607. if app_path is None:
  608. app_path = oozie_workflow.appPath
  609. return_json = request.GET.get('format') == 'json'
  610. if request.method == 'POST':
  611. rerun_form = RerunForm(request.POST, oozie_workflow=oozie_workflow)
  612. params_form = ParametersFormSet(request.POST)
  613. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  614. args = {}
  615. if request.POST['rerun_form_choice'] == 'fail_nodes':
  616. args['fail_nodes'] = 'true'
  617. else:
  618. args['skip_nodes'] = ','.join(rerun_form.cleaned_data['skip_nodes'])
  619. args['deployment_dir'] = app_path
  620. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  621. _rerun_workflow(request, job_id, args, mapping)
  622. if rerun_form.cleaned_data['return_json']:
  623. return JsonResponse({'status': 0, 'job_id': job_id}, safe=False)
  624. else:
  625. request.info(_('Workflow re-running.'))
  626. return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_id}))
  627. else:
  628. request.error(_('Invalid submission form: %s %s' % (rerun_form.errors, params_form.errors)))
  629. else:
  630. rerun_form = RerunForm(oozie_workflow=oozie_workflow, return_json=return_json)
  631. initial_params = ParameterForm.get_initial_params(oozie_workflow.conf_dict)
  632. params_form = ParametersFormSet(initial=initial_params)
  633. popup = render('dashboard/rerun_workflow_popup.mako', request, {
  634. 'rerun_form': rerun_form,
  635. 'params_form': params_form,
  636. 'action': reverse('oozie:rerun_oozie_job', kwargs={'job_id': job_id, 'app_path': app_path}),
  637. 'return_json': return_json
  638. }, force_template=True).content
  639. return JsonResponse(popup, safe=False)
  640. def _rerun_workflow(request, oozie_id, run_args, mapping):
  641. try:
  642. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, properties=mapping, oozie_id=oozie_id)
  643. job_id = submission.rerun(**run_args)
  644. return job_id
  645. except RestException, ex:
  646. msg = _("Error re-running workflow %s.") % (oozie_id,)
  647. LOG.exception(msg)
  648. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  649. @show_oozie_error
  650. def rerun_oozie_coordinator(request, job_id, app_path=None):
  651. oozie_coordinator = check_job_access_permission(request, job_id)
  652. check_job_edition_permission(oozie_coordinator, request.user)
  653. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  654. if app_path is None:
  655. app_path = oozie_coordinator.coordJobPath
  656. return_json = request.GET.get('format') == 'json'
  657. if request.method == 'POST':
  658. params_form = ParametersFormSet(request.POST)
  659. rerun_form = RerunCoordForm(request.POST, oozie_coordinator=oozie_coordinator)
  660. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  661. args = {}
  662. args['deployment_dir'] = app_path
  663. params = {
  664. 'type': 'action',
  665. 'scope': ','.join(oozie_coordinator.aggreate(rerun_form.cleaned_data['actions'])),
  666. 'refresh': rerun_form.cleaned_data['refresh'],
  667. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  668. }
  669. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  670. _rerun_coordinator(request, job_id, args, params, properties)
  671. if rerun_form.cleaned_data['return_json']:
  672. return JsonResponse({'status': 0, 'job_id': job_id}, safe=False)
  673. else:
  674. request.info(_('Coordinator re-running.'))
  675. return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
  676. else:
  677. request.error(_('Invalid submission form: %s') % smart_unicode(rerun_form.errors))
  678. return list_oozie_coordinator(request, job_id)
  679. else:
  680. rerun_form = RerunCoordForm(oozie_coordinator=oozie_coordinator, return_json=return_json)
  681. initial_params = ParameterForm.get_initial_params(oozie_coordinator.conf_dict)
  682. params_form = ParametersFormSet(initial=initial_params)
  683. popup = render('dashboard/rerun_coord_popup.mako', request, {
  684. 'rerun_form': rerun_form,
  685. 'params_form': params_form,
  686. 'action': reverse('oozie:rerun_oozie_coord', kwargs={'job_id': job_id, 'app_path': app_path}),
  687. 'return_json': return_json,
  688. }, force_template=True).content
  689. return JsonResponse(popup, safe=False)
  690. def _rerun_coordinator(request, oozie_id, args, params, properties):
  691. try:
  692. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  693. job_id = submission.rerun_coord(params=params, **args)
  694. return job_id
  695. except RestException, ex:
  696. msg = _("Error re-running coordinator %s.") % (oozie_id,)
  697. LOG.exception(msg)
  698. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  699. @show_oozie_error
  700. def rerun_oozie_bundle(request, job_id, app_path):
  701. oozie_bundle = check_job_access_permission(request, job_id)
  702. check_job_edition_permission(oozie_bundle, request.user)
  703. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  704. if request.method == 'POST':
  705. params_form = ParametersFormSet(request.POST)
  706. rerun_form = RerunBundleForm(request.POST, oozie_bundle=oozie_bundle)
  707. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  708. args = {}
  709. args['deployment_dir'] = app_path
  710. params = {
  711. 'coord-scope': ','.join(rerun_form.cleaned_data['coordinators']),
  712. 'refresh': rerun_form.cleaned_data['refresh'],
  713. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  714. }
  715. if rerun_form.cleaned_data['start'] and rerun_form.cleaned_data['end']:
  716. date = {
  717. 'date-scope':
  718. '%(start)s::%(end)s' % {
  719. 'start': utc_datetime_format(rerun_form.cleaned_data['start']),
  720. 'end': utc_datetime_format(rerun_form.cleaned_data['end'])
  721. }
  722. }
  723. params.update(date)
  724. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  725. _rerun_bundle(request, job_id, args, params, properties)
  726. request.info(_('Bundle re-running.'))
  727. return redirect(reverse('oozie:list_oozie_bundle', kwargs={'job_id': job_id}))
  728. else:
  729. request.error(_('Invalid submission form: %s' % (rerun_form.errors,)))
  730. return list_oozie_bundle(request, job_id)
  731. else:
  732. rerun_form = RerunBundleForm(oozie_bundle=oozie_bundle)
  733. initial_params = ParameterForm.get_initial_params(oozie_bundle.conf_dict)
  734. params_form = ParametersFormSet(initial=initial_params)
  735. popup = render('dashboard/rerun_bundle_popup.mako', request, {
  736. 'rerun_form': rerun_form,
  737. 'params_form': params_form,
  738. 'action': reverse('oozie:rerun_oozie_bundle', kwargs={'job_id': job_id, 'app_path': app_path}),
  739. }, force_template=True).content
  740. return JsonResponse(popup, safe=False)
  741. def _rerun_bundle(request, oozie_id, args, params, properties):
  742. try:
  743. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  744. job_id = submission.rerun_bundle(params=params, **args)
  745. return job_id
  746. except RestException, ex:
  747. msg = _("Error re-running bundle %s.") % (oozie_id,)
  748. LOG.exception(msg)
  749. raise PopupException(msg, detail=ex._headers.get('oozie-error-message', ex))
  750. def submit_external_job(request, application_path):
  751. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  752. if request.method == 'POST':
  753. params_form = ParametersFormSet(request.POST)
  754. if params_form.is_valid():
  755. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  756. mapping['dryrun'] = request.POST.get('dryrun_checkbox') == 'on'
  757. application_name = os.path.basename(application_path)
  758. application_class = Bundle if application_name == 'bundle.xml' else Coordinator if application_name == 'coordinator.xml' else get_workflow()
  759. mapping[application_class.get_application_path_key()] = os.path.dirname(application_path)
  760. try:
  761. submission = Submission(request.user, fs=request.fs, jt=request.jt, properties=mapping)
  762. job_id = submission.run(application_path)
  763. except RestException, ex:
  764. detail = ex._headers.get('oozie-error-message', ex)
  765. if 'Max retries exceeded with url' in str(detail):
  766. detail = '%s: %s' % (_('The Oozie server is not running'), detail)
  767. LOG.exception(smart_str(detail))
  768. raise PopupException(_("Error submitting job %s") % (application_path,), detail=detail)
  769. jsonify = request.POST.get('format') == 'json'
  770. if jsonify:
  771. return JsonResponse({'status': 0, 'job_id': job_id, 'type': 'external_workflow'}, safe=False)
  772. else:
  773. request.info(_('Oozie job submitted'))
  774. view = 'list_oozie_bundle' if application_name == 'bundle.xml' else 'list_oozie_coordinator' if application_name == 'coordinator.xml' else 'list_oozie_workflow'
  775. return redirect(reverse('oozie:%s' % view, kwargs={'job_id': job_id}))
  776. else:
  777. request.error(_('Invalid submission form: %s' % params_form.errors))
  778. else:
  779. parameters = Submission(request.user, fs=request.fs, jt=request.jt).get_external_parameters(application_path)
  780. initial_params = ParameterForm.get_initial_params(parameters)
  781. params_form = ParametersFormSet(initial=initial_params)
  782. popup = render('editor/submit_job_popup.mako', request, {
  783. 'params_form': params_form,
  784. 'name': _('Job'),
  785. 'action': reverse('oozie:submit_external_job', kwargs={'application_path': application_path}),
  786. 'show_dryrun': os.path.basename(application_path) != 'bundle.xml',
  787. 'return_json': request.GET.get('format') == 'json'
  788. }, force_template=True).content
  789. return JsonResponse(popup, safe=False)
  790. def massaged_workflow_actions_for_json(workflow_actions, oozie_coordinator, oozie_bundle):
  791. actions = []
  792. for action in workflow_actions:
  793. if oozie_coordinator is not None:
  794. setattr(action, 'oozie_coordinator', oozie_coordinator)
  795. if oozie_bundle is not None:
  796. setattr(action, 'oozie_bundle', oozie_bundle)
  797. massaged_action = {
  798. 'id': action.id,
  799. 'log': action.get_absolute_log_url(),
  800. 'url': action.get_absolute_url(),
  801. 'name': action.name,
  802. 'type': action.type,
  803. 'status': action.status,
  804. 'externalIdUrl': action.get_external_id_url(),
  805. 'externalId': action.externalId,
  806. 'startTime': format_time(action.startTime),
  807. 'endTime': format_time(action.endTime),
  808. 'retries': action.retries,
  809. 'errorCode': action.errorCode,
  810. 'errorMessage': action.errorMessage,
  811. 'transition': action.transition,
  812. 'data': action.data,
  813. }
  814. actions.append(massaged_action)
  815. return actions
  816. def massaged_coordinator_actions_for_json(coordinator, oozie_bundle):
  817. coordinator_id = coordinator.id
  818. coordinator_actions = coordinator.get_working_actions()
  819. actions = []
  820. related_job_ids = []
  821. related_job_ids.append('coordinator_job_id=%s' % coordinator_id)
  822. if oozie_bundle is not None:
  823. related_job_ids.append('bundle_job_id=%s' %oozie_bundle.id)
  824. for action in coordinator_actions:
  825. massaged_action = {
  826. 'id': action.id,
  827. 'url': action.externalId and reverse('oozie:list_oozie_workflow', kwargs={'job_id': action.externalId}) + '?%s' % '&'.join(related_job_ids) or '',
  828. 'number': action.actionNumber,
  829. 'type': 'schedule-task',
  830. 'status': action.status,
  831. 'externalId': action.externalId or '-',
  832. 'externalIdUrl': action.externalId and reverse('oozie:list_oozie_workflow_action', kwargs={'action': action.externalId}) or '',
  833. 'nominalTime': format_time(action.nominalTime),
  834. 'title': action.title,
  835. 'createdTime': format_time(action.createdTime),
  836. 'lastModifiedTime': format_time(action.lastModifiedTime),
  837. 'errorCode': action.errorCode,
  838. 'errorMessage': action.errorMessage,
  839. 'missingDependencies': action.missingDependencies
  840. }
  841. actions.append(massaged_action)
  842. # Sorting for Oozie < 4.1 backward compatibility
  843. actions.sort(key=lambda k: k['number'], reverse=True)
  844. return actions
  845. def massaged_bundle_actions_for_json(bundle):
  846. bundle_actions = bundle.get_working_actions()
  847. actions = []
  848. for action in bundle_actions:
  849. massaged_action = {
  850. 'id': action.coordJobId,
  851. 'url': action.coordJobId and reverse('oozie:list_oozie_coordinator', kwargs={'job_id': action.coordJobId}) + '?bundle_job_id=%s' % bundle.id or '',
  852. 'name': action.coordJobName,
  853. 'type': action.type,
  854. 'status': action.status,
  855. 'externalId': action.coordExternalId or '-',
  856. 'frequency': action.frequency,
  857. 'timeUnit': action.timeUnit,
  858. 'nextMaterializedTime': action.nextMaterializedTime,
  859. 'concurrency': action.concurrency,
  860. 'pauseTime': action.pauseTime,
  861. 'user': action.user,
  862. 'acl': action.acl,
  863. 'timeOut': action.timeOut,
  864. 'coordJobPath': action.coordJobPath,
  865. 'executionPolicy': action.executionPolicy,
  866. 'startTime': action.startTime,
  867. 'endTime': action.endTime,
  868. 'lastAction': action.lastAction
  869. }
  870. actions.insert(0, massaged_action)
  871. return actions
  872. def format_time(st_time):
  873. if st_time is None:
  874. return '-'
  875. elif type(st_time) == time.struct_time:
  876. return time.strftime("%a, %d %b %Y %H:%M:%S", st_time)
  877. else:
  878. return st_time
  879. def massaged_oozie_jobs_for_json(oozie_jobs, user, just_sla=False):
  880. jobs = []
  881. for job in oozie_jobs:
  882. if not just_sla or (just_sla and job.has_sla) and job.appName != 'pig-app-hue-script':
  883. last_modified_time_millis = hasattr(job, 'lastModTime') and job.lastModTime and (time.time() - time.mktime(job.lastModTime)) * 1000 or 0
  884. duration_millis = job.durationTime
  885. massaged_job = {
  886. 'id': job.id,
  887. 'lastModTime': hasattr(job, 'lastModTime') and job.lastModTime and format_time(job.lastModTime) or None,
  888. 'lastModTimeInMillis': last_modified_time_millis,
  889. 'lastModTimeFormatted': last_modified_time_millis and format_duration_in_millis(last_modified_time_millis) or None,
  890. 'kickoffTime': hasattr(job, 'kickoffTime') and job.kickoffTime and format_time(job.kickoffTime) or '',
  891. 'kickoffTimeInMillis': hasattr(job, 'kickoffTime') and job.kickoffTime and time.mktime(catch_unicode_time(job.kickoffTime)) or 0,
  892. 'nextMaterializedTime': hasattr(job, 'nextMaterializedTime') and job.nextMaterializedTime and format_time(job.nextMaterializedTime) or '',
  893. 'nextMaterializedTimeInMillis': hasattr(job, 'nextMaterializedTime') and job.nextMaterializedTime and time.mktime(job.nextMaterializedTime) or 0,
  894. 'timeOut': hasattr(job, 'timeOut') and job.timeOut or None,
  895. 'endTime': job.endTime and format_time(job.endTime) or None,
  896. 'pauseTime': hasattr(job, 'pauseTime') and job.pauseTime and format_time(job.endTime) or None,
  897. 'concurrency': hasattr(job, 'concurrency') and job.concurrency or None,
  898. 'endTimeInMillis': job.endTime and time.mktime(job.endTime) or 0,
  899. 'lastActionInMillis': hasattr(job, 'lastAction') and job.lastAction and time.mktime(job.lastAction) or 0,
  900. 'status': job.status,
  901. 'group': job.group,
  902. 'isRunning': job.is_running(),
  903. 'duration': duration_millis and format_duration_in_millis(duration_millis) or None,
  904. 'durationInMillis': duration_millis,
  905. 'appName': job.appName,
  906. 'progress': job.get_progress(),
  907. 'user': job.user,
  908. 'absoluteUrl': job.get_absolute_url(),
  909. 'canEdit': has_job_edition_permission(job, user),
  910. 'killUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'kill'}),
  911. 'suspendUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'suspend'}),
  912. 'resumeUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'resume'}),
  913. 'created': hasattr(job, 'createdTime') and job.createdTime and format_time(job.createdTime) or '',
  914. 'createdInMillis': job.submissionTime,
  915. 'startTime': hasattr(job, 'startTime') and format_time(job.startTime) or None,
  916. 'startTimeInMillis': hasattr(job, 'startTime') and job.startTime and time.mktime(job.startTime) or 0,
  917. 'run': hasattr(job, 'run') and job.run or 0,
  918. 'frequency': hasattr(job, 'frequency') and Coordinator.CRON_MAPPING.get(job.frequency, job.frequency) or None,
  919. 'timeUnit': hasattr(job, 'timeUnit') and job.timeUnit or None,
  920. 'parentUrl': hasattr(job, 'parentId') and job.parentId and get_link(job.parentId) or '',
  921. 'submittedManually': hasattr(job, 'parentId') and (job.parentId is None or 'C@' not in job.parentId)
  922. }
  923. jobs.append(massaged_job)
  924. return { 'jobs': jobs }
  925. def check_job_access_permission(request, job_id, **kwargs):
  926. """
  927. Decorator ensuring that the user has access to the job submitted to Oozie.
  928. Arg: Oozie 'workflow', 'coordinator' or 'bundle' ID.
  929. Return: the Oozie workflow, coordinator or bundle or raise an exception
  930. Notice: its gets an id in input and returns the full object in output (not an id).
  931. """
  932. if job_id is not None:
  933. oozie_api = get_oozie(request.user)
  934. if job_id.endswith('W'):
  935. get_job = oozie_api.get_job
  936. elif job_id.endswith('C'):
  937. get_job = oozie_api.get_coordinator
  938. else:
  939. get_job = oozie_api.get_bundle
  940. try:
  941. if job_id.endswith('C'):
  942. oozie_job = get_job(job_id, **kwargs)
  943. else:
  944. oozie_job = get_job(job_id)
  945. except RestException, ex:
  946. msg = _("Error accessing Oozie job %s.") % (job_id,)
  947. LOG.exception(msg)
  948. raise PopupException(msg, detail=ex._headers.get('oozie-error-message'))
  949. if request.user.is_superuser \
  950. or oozie_job.user == request.user.username \
  951. or has_dashboard_jobs_access(request.user):
  952. return oozie_job
  953. else:
  954. message = _("Permission denied. %(username)s does not have the permissions to access job %(id)s.") % \
  955. {'username': request.user.username, 'id': oozie_job.id}
  956. access_warn(request, message)
  957. raise PopupException(message)
  958. def check_job_edition_permission(oozie_job, user):
  959. if has_job_edition_permission(oozie_job, user):
  960. return oozie_job
  961. else:
  962. message = _("Permission denied. %(username)s does not have the permissions to modify job %(id)s.") % \
  963. {'username': user.username, 'id': oozie_job.id}
  964. raise PopupException(message)
  965. def has_job_edition_permission(oozie_job, user):
  966. 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(','))
  967. def has_dashboard_jobs_access(user):
  968. return user.is_superuser or user.has_hue_permission(action="dashboard_jobs_access", app=DJANGO_APPS[0])