dashboard.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866
  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.lib.django_util import render, encode_json_for_js
  29. from desktop.lib.exceptions_renderable import PopupException
  30. from desktop.lib.i18n import smart_str
  31. from desktop.lib.rest.http_client import RestException
  32. from desktop.lib.view_util import format_duration_in_millis
  33. from desktop.log.access import access_warn
  34. from liboozie.oozie_api import get_oozie
  35. from liboozie.submittion import Submission
  36. from liboozie.types import Workflow as OozieWorkflow
  37. from oozie.conf import OOZIE_JOBS_COUNT, ENABLE_CRON_SCHEDULING
  38. from oozie.forms import RerunForm, ParameterForm, RerunCoordForm,\
  39. RerunBundleForm
  40. from oozie.models import History, Job, Workflow, utc_datetime_format, Bundle,\
  41. Coordinator, get_link
  42. from oozie.settings import DJANGO_APPS
  43. LOG = logging.getLogger(__name__)
  44. MAX_COORD_ACTIONS = 250
  45. """
  46. Permissions:
  47. A Workflow/Coordinator/Bundle can:
  48. * be accessed only by its owner or a superuser or by a user with 'dashboard_jobs_access' permissions
  49. * be submitted/modified only by its owner or a superuser
  50. Permissions checking happens by calling:
  51. * check_job_access_permission()
  52. * check_job_edition_permission()
  53. """
  54. def manage_oozie_jobs(request, job_id, action):
  55. if request.method != 'POST':
  56. raise PopupException(_('Use a POST request to manage an Oozie job.'))
  57. job = check_job_access_permission(request, job_id)
  58. check_job_edition_permission(job, request.user)
  59. response = {'status': -1, 'data': ''}
  60. try:
  61. response['data'] = get_oozie(request.user).job_control(job_id, action)
  62. response['status'] = 0
  63. if 'notification' in request.POST:
  64. request.info(_(request.POST.get('notification')))
  65. except RestException, ex:
  66. response['data'] = _("Error performing %s on Oozie job %s: %s.") % (action, job_id, ex.message)
  67. return HttpResponse(json.dumps(response), mimetype="application/json")
  68. def show_oozie_error(view_func):
  69. def decorate(request, *args, **kwargs):
  70. try:
  71. return view_func(request, *args, **kwargs)
  72. except RestException, ex:
  73. detail = ex._headers.get('oozie-error-message', ex)
  74. if 'Max retries exceeded with url' in str(detail):
  75. detail = '%s: %s' % (_('The Oozie server is not running'), detail)
  76. raise PopupException(_('An error occurred with Oozie.'), detail=detail)
  77. return wraps(view_func)(decorate)
  78. @show_oozie_error
  79. def list_oozie_workflows(request):
  80. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(),}
  81. if not has_dashboard_jobs_access(request.user):
  82. kwargs['user'] = request.user.username
  83. if request.GET.get('format') == 'json':
  84. just_sla = request.GET.get('justsla') == 'true'
  85. if request.GET.get('type') == 'running':
  86. kwargs['filters'] = [('status', status) for status in OozieWorkflow.RUNNING_STATUSES]
  87. json_jobs = get_oozie(request.user).get_workflows(**kwargs).jobs
  88. elif request.GET.get('type') == 'completed':
  89. kwargs['filters'] = [('status', status) for status in OozieWorkflow.FINISHED_STATUSES]
  90. json_jobs = get_oozie(request.user).get_workflows(**kwargs).jobs
  91. elif request.GET.get('type') == 'progress':
  92. kwargs['filters'] = [('status', status) for status in OozieWorkflow.RUNNING_STATUSES]
  93. json_jobs = get_oozie(request.user).get_workflows(**kwargs).jobs
  94. json_jobs = [get_oozie(request.user).get_job(job.id) for job in json_jobs]
  95. return HttpResponse(encode_json_for_js(massaged_oozie_jobs_for_json(json_jobs, request.user, just_sla)), mimetype="application/json")
  96. return render('dashboard/list_oozie_workflows.mako', request, {
  97. 'user': request.user,
  98. 'jobs': [],
  99. 'has_job_edition_permission': has_job_edition_permission,
  100. })
  101. @show_oozie_error
  102. def list_oozie_coordinators(request):
  103. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(),}
  104. if not has_dashboard_jobs_access(request.user):
  105. kwargs['user'] = request.user.username
  106. coordinators = get_oozie(request.user).get_coordinators(**kwargs)
  107. enable_cron_scheduling = ENABLE_CRON_SCHEDULING.get()
  108. if request.GET.get('format') == 'json':
  109. json_jobs = coordinators.jobs
  110. if request.GET.get('type') == 'running':
  111. json_jobs = split_oozie_jobs(request.user, coordinators.jobs)['running_jobs']
  112. if request.GET.get('type') == 'completed':
  113. json_jobs = split_oozie_jobs(request.user, coordinators.jobs)['completed_jobs']
  114. return HttpResponse(json.dumps(massaged_oozie_jobs_for_json(json_jobs, request.user)).replace('\\\\', '\\'), mimetype="application/json")
  115. return render('dashboard/list_oozie_coordinators.mako', request, {
  116. 'jobs': split_oozie_jobs(request.user, coordinators.jobs),
  117. 'has_job_edition_permission': has_job_edition_permission,
  118. 'enable_cron_scheduling': enable_cron_scheduling,
  119. })
  120. @show_oozie_error
  121. def list_oozie_bundles(request):
  122. kwargs = {'cnt': OOZIE_JOBS_COUNT.get(),}
  123. if not has_dashboard_jobs_access(request.user):
  124. kwargs['user'] = request.user.username
  125. bundles = get_oozie(request.user).get_bundles(**kwargs)
  126. if request.GET.get('format') == 'json':
  127. json_jobs = bundles.jobs
  128. if request.GET.get('type') == 'running':
  129. json_jobs = split_oozie_jobs(request.user, bundles.jobs)['running_jobs']
  130. if request.GET.get('type') == 'completed':
  131. json_jobs = split_oozie_jobs(request.user, bundles.jobs)['completed_jobs']
  132. return HttpResponse(json.dumps(massaged_oozie_jobs_for_json(json_jobs, request.user)).replace('\\\\', '\\'), mimetype="application/json")
  133. return render('dashboard/list_oozie_bundles.mako', request, {
  134. 'jobs': split_oozie_jobs(request.user, bundles.jobs),
  135. 'has_job_edition_permission': has_job_edition_permission,
  136. })
  137. @show_oozie_error
  138. def list_oozie_workflow(request, job_id):
  139. oozie_workflow = check_job_access_permission(request, job_id)
  140. oozie_coordinator = None
  141. if request.GET.get('coordinator_job_id'):
  142. oozie_coordinator = check_job_access_permission(request, request.GET.get('coordinator_job_id'))
  143. oozie_bundle = None
  144. if request.GET.get('bundle_job_id'):
  145. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  146. if oozie_coordinator is not None:
  147. setattr(oozie_workflow, 'oozie_coordinator', oozie_coordinator)
  148. if oozie_bundle is not None:
  149. setattr(oozie_workflow, 'oozie_bundle', oozie_bundle)
  150. history = History.cross_reference_submission_history(request.user, job_id)
  151. hue_coord = history and history.get_coordinator() or History.get_coordinator_from_config(oozie_workflow.conf_dict)
  152. hue_workflow = (hue_coord and hue_coord.workflow) or (history and history.get_workflow()) or History.get_workflow_from_config(oozie_workflow.conf_dict)
  153. if hue_coord and hue_coord.workflow: Job.objects.can_read_or_exception(request, hue_coord.workflow.id)
  154. if hue_workflow: Job.objects.can_read_or_exception(request, hue_workflow.id)
  155. parameters = oozie_workflow.conf_dict.copy()
  156. for action in oozie_workflow.actions:
  157. action.oozie_coordinator = oozie_coordinator
  158. action.oozie_bundle = oozie_bundle
  159. if hue_workflow:
  160. workflow_graph = hue_workflow.gen_status_graph(oozie_workflow)
  161. full_node_list = hue_workflow.node_list
  162. else:
  163. workflow_graph, full_node_list = Workflow.gen_status_graph_from_xml(request.user, oozie_workflow)
  164. if request.GET.get('format') == 'json':
  165. return_obj = {
  166. 'id': oozie_workflow.id,
  167. 'status': oozie_workflow.status,
  168. 'progress': oozie_workflow.get_progress(full_node_list),
  169. 'graph': workflow_graph,
  170. 'actions': massaged_workflow_actions_for_json(oozie_workflow.get_working_actions(), oozie_coordinator, oozie_bundle)
  171. }
  172. return HttpResponse(encode_json_for_js(return_obj), mimetype="application/json")
  173. oozie_slas = []
  174. if oozie_workflow.has_sla:
  175. api = get_oozie(request.user, api_version="v2")
  176. params = {
  177. 'id': oozie_workflow.id,
  178. 'parent_id': oozie_workflow.id
  179. }
  180. oozie_slas = api.get_oozie_slas(**params)
  181. return render('dashboard/list_oozie_workflow.mako', request, {
  182. 'history': history,
  183. 'oozie_workflow': oozie_workflow,
  184. 'oozie_coordinator': oozie_coordinator,
  185. 'oozie_bundle': oozie_bundle,
  186. 'oozie_slas': oozie_slas,
  187. 'hue_workflow': hue_workflow,
  188. 'hue_coord': hue_coord,
  189. 'parameters': parameters,
  190. 'has_job_edition_permission': has_job_edition_permission,
  191. 'workflow_graph': workflow_graph
  192. })
  193. @show_oozie_error
  194. def list_oozie_coordinator(request, job_id):
  195. oozie_coordinator = check_job_access_permission(request, job_id)
  196. # Cross reference the submission history (if any)
  197. coordinator = None
  198. try:
  199. coordinator = History.objects.get(oozie_job_id=job_id).job.get_full_node()
  200. except History.DoesNotExist:
  201. pass
  202. oozie_bundle = None
  203. if request.GET.get('bundle_job_id'):
  204. try:
  205. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  206. except:
  207. pass
  208. show_all_actions =request.GET.get('show_all_actions') == 'true'
  209. if request.GET.get('format') == 'json':
  210. actions = massaged_coordinator_actions_for_json(oozie_coordinator, oozie_bundle)
  211. if not show_all_actions:
  212. actions = actions[:MAX_COORD_ACTIONS]
  213. return_obj = {
  214. 'id': oozie_coordinator.id,
  215. 'status': oozie_coordinator.status,
  216. 'progress': oozie_coordinator.get_progress(),
  217. 'nextTime': format_time(oozie_coordinator.nextMaterializedTime),
  218. 'endTime': format_time(oozie_coordinator.endTime),
  219. 'actions': actions,
  220. 'show_all_actions': show_all_actions
  221. }
  222. return HttpResponse(encode_json_for_js(return_obj), mimetype="application/json")
  223. oozie_slas = []
  224. if oozie_coordinator.has_sla:
  225. api = get_oozie(request.user, api_version="v2")
  226. params = {
  227. 'id': oozie_coordinator.id,
  228. 'parent_id': oozie_coordinator.id
  229. }
  230. oozie_slas = api.get_oozie_slas(**params)
  231. enable_cron_scheduling = ENABLE_CRON_SCHEDULING.get()
  232. return render('dashboard/list_oozie_coordinator.mako', request, {
  233. 'oozie_coordinator': oozie_coordinator,
  234. 'oozie_slas': oozie_slas,
  235. 'coordinator': coordinator,
  236. 'oozie_bundle': oozie_bundle,
  237. 'has_job_edition_permission': has_job_edition_permission,
  238. 'show_all_actions': show_all_actions,
  239. 'MAX_COORD_ACTIONS': MAX_COORD_ACTIONS,
  240. 'enable_cron_scheduling': enable_cron_scheduling,
  241. })
  242. @show_oozie_error
  243. def list_oozie_bundle(request, job_id):
  244. oozie_bundle = check_job_access_permission(request, job_id)
  245. # Cross reference the submission history (if any)
  246. bundle = None
  247. try:
  248. bundle = History.objects.get(oozie_job_id=job_id).job.get_full_node()
  249. except History.DoesNotExist:
  250. pass
  251. if request.GET.get('format') == 'json':
  252. return_obj = {
  253. 'id': oozie_bundle.id,
  254. 'status': oozie_bundle.status,
  255. 'progress': oozie_bundle.get_progress(),
  256. 'endTime': format_time(oozie_bundle.endTime),
  257. 'actions': massaged_bundle_actions_for_json(oozie_bundle)
  258. }
  259. return HttpResponse(json.dumps(return_obj).replace('\\\\', '\\'), mimetype="application/json")
  260. return render('dashboard/list_oozie_bundle.mako', request, {
  261. 'oozie_bundle': oozie_bundle,
  262. 'bundle': bundle,
  263. 'has_job_edition_permission': has_job_edition_permission,
  264. })
  265. @show_oozie_error
  266. def list_oozie_workflow_action(request, action):
  267. try:
  268. action = get_oozie(request.user).get_action(action)
  269. workflow = check_job_access_permission(request, action.id.split('@')[0])
  270. except RestException, ex:
  271. raise PopupException(_("Error accessing Oozie action %s.") % (action,), detail=ex.message)
  272. oozie_coordinator = None
  273. if request.GET.get('coordinator_job_id'):
  274. oozie_coordinator = check_job_access_permission(request, request.GET.get('coordinator_job_id'))
  275. oozie_bundle = None
  276. if request.GET.get('bundle_job_id'):
  277. oozie_bundle = check_job_access_permission(request, request.GET.get('bundle_job_id'))
  278. workflow.oozie_coordinator = oozie_coordinator
  279. workflow.oozie_bundle = oozie_bundle
  280. return render('dashboard/list_oozie_workflow_action.mako', request, {
  281. 'action': action,
  282. 'workflow': workflow,
  283. 'oozie_coordinator': oozie_coordinator,
  284. 'oozie_bundle': oozie_bundle,
  285. })
  286. @show_oozie_error
  287. def get_oozie_job_log(request, job_id):
  288. oozie_job = check_job_access_permission(request, job_id)
  289. return_obj = {
  290. 'id': oozie_job.id,
  291. 'status': oozie_job.status,
  292. 'log': oozie_job.log,
  293. }
  294. return HttpResponse(encode_json_for_js(return_obj), mimetype="application/json")
  295. @show_oozie_error
  296. def list_oozie_info(request):
  297. api = get_oozie(request.user)
  298. instrumentation = api.get_instrumentation()
  299. configuration = api.get_configuration()
  300. oozie_status = api.get_oozie_status()
  301. return render('dashboard/list_oozie_info.mako', request, {
  302. 'instrumentation': instrumentation,
  303. 'configuration': configuration,
  304. 'oozie_status': oozie_status,
  305. })
  306. @show_oozie_error
  307. def list_oozie_sla(request):
  308. api = get_oozie(request.user, api_version="v2")
  309. if request.method == 'POST':
  310. params = {}
  311. job_name = request.POST.get('job_name')
  312. if re.match('.*-oozie-oozi-[WCB]', job_name):
  313. params['id'] = job_name
  314. params['parent_id'] = job_name
  315. else:
  316. params['app_name'] = job_name
  317. if 'useDates' in request.POST:
  318. if request.POST.get('start'):
  319. params['nominal_start'] = request.POST.get('start')
  320. if request.POST.get('end'):
  321. params['nominal_end'] = request.POST.get('end')
  322. oozie_slas = api.get_oozie_slas(**params)
  323. else:
  324. oozie_slas = [] # or get latest?
  325. if request.REQUEST.get('format') == 'json':
  326. massaged_slas = []
  327. for sla in oozie_slas:
  328. massaged_slas.append(massaged_sla_for_json(sla, request))
  329. return HttpResponse(json.dumps({'oozie_slas': massaged_slas}), content_type="text/json")
  330. return render('dashboard/list_oozie_sla.mako', request, {
  331. 'oozie_slas': oozie_slas
  332. })
  333. def massaged_sla_for_json(sla, request):
  334. massaged_sla = {
  335. 'slaStatus': sla['slaStatus'],
  336. 'id': sla['id'],
  337. 'appType': sla['appType'],
  338. 'appName': sla['appName'],
  339. 'appUrl': get_link(sla['id']),
  340. 'user': sla['user'],
  341. 'nominalTime': sla['nominalTime'],
  342. 'expectedStart': sla['expectedStart'],
  343. 'actualStart': sla['actualStart'],
  344. 'expectedEnd': sla['expectedEnd'],
  345. 'actualEnd': sla['actualEnd'],
  346. 'jobStatus': sla['jobStatus'],
  347. 'expectedDuration': sla['expectedDuration'],
  348. 'actualDuration': sla['actualDuration'],
  349. 'lastModified': sla['lastModified']
  350. }
  351. return massaged_sla
  352. @show_oozie_error
  353. def rerun_oozie_job(request, job_id, app_path):
  354. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  355. oozie_workflow = check_job_access_permission(request, job_id)
  356. check_job_edition_permission(oozie_workflow, request.user)
  357. if request.method == 'POST':
  358. rerun_form = RerunForm(request.POST, oozie_workflow=oozie_workflow)
  359. params_form = ParametersFormSet(request.POST)
  360. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  361. args = {}
  362. if request.POST['rerun_form_choice'] == 'fail_nodes':
  363. args['fail_nodes'] = 'true'
  364. else:
  365. args['skip_nodes'] = ','.join(rerun_form.cleaned_data['skip_nodes'])
  366. args['deployment_dir'] = app_path
  367. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  368. _rerun_workflow(request, job_id, args, mapping)
  369. request.info(_('Workflow re-running.'))
  370. return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_id}))
  371. else:
  372. request.error(_('Invalid submission form: %s %s' % (rerun_form.errors, params_form.errors)))
  373. else:
  374. rerun_form = RerunForm(oozie_workflow=oozie_workflow)
  375. initial_params = ParameterForm.get_initial_params(oozie_workflow.conf_dict)
  376. params_form = ParametersFormSet(initial=initial_params)
  377. popup = render('dashboard/rerun_job_popup.mako', request, {
  378. 'rerun_form': rerun_form,
  379. 'params_form': params_form,
  380. 'action': reverse('oozie:rerun_oozie_job', kwargs={'job_id': job_id, 'app_path': app_path}),
  381. }, force_template=True).content
  382. return HttpResponse(json.dumps(popup), mimetype="application/json")
  383. def _rerun_workflow(request, oozie_id, run_args, mapping):
  384. try:
  385. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, properties=mapping, oozie_id=oozie_id)
  386. job_id = submission.rerun(**run_args)
  387. return job_id
  388. except RestException, ex:
  389. raise PopupException(_("Error re-running workflow %s.") % (oozie_id,),
  390. detail=ex._headers.get('oozie-error-message', ex))
  391. @show_oozie_error
  392. def rerun_oozie_coordinator(request, job_id, app_path):
  393. oozie_coordinator = check_job_access_permission(request, job_id)
  394. check_job_edition_permission(oozie_coordinator, request.user)
  395. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  396. if request.method == 'POST':
  397. params_form = ParametersFormSet(request.POST)
  398. rerun_form = RerunCoordForm(request.POST, oozie_coordinator=oozie_coordinator)
  399. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  400. args = {}
  401. args['deployment_dir'] = app_path
  402. params = {
  403. 'type': 'action',
  404. 'scope': ','.join(oozie_coordinator.aggreate(rerun_form.cleaned_data['actions'])),
  405. 'refresh': rerun_form.cleaned_data['refresh'],
  406. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  407. }
  408. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  409. _rerun_coordinator(request, job_id, args, params, properties)
  410. request.info(_('Coordinator re-running.'))
  411. return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
  412. else:
  413. request.error(_('Invalid submission form: %s' % (rerun_form.errors,)))
  414. return list_oozie_coordinator(request, job_id)
  415. else:
  416. rerun_form = RerunCoordForm(oozie_coordinator=oozie_coordinator)
  417. initial_params = ParameterForm.get_initial_params(oozie_coordinator.conf_dict)
  418. params_form = ParametersFormSet(initial=initial_params)
  419. popup = render('dashboard/rerun_coord_popup.mako', request, {
  420. 'rerun_form': rerun_form,
  421. 'params_form': params_form,
  422. 'action': reverse('oozie:rerun_oozie_coord', kwargs={'job_id': job_id, 'app_path': app_path}),
  423. }, force_template=True).content
  424. return HttpResponse(json.dumps(popup), mimetype="application/json")
  425. def _rerun_coordinator(request, oozie_id, args, params, properties):
  426. try:
  427. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  428. job_id = submission.rerun_coord(params=params, **args)
  429. return job_id
  430. except RestException, ex:
  431. raise PopupException(_("Error re-running coordinator %s.") % (oozie_id,),
  432. detail=ex._headers.get('oozie-error-message', ex))
  433. @show_oozie_error
  434. def rerun_oozie_bundle(request, job_id, app_path):
  435. oozie_bundle = check_job_access_permission(request, job_id)
  436. check_job_edition_permission(oozie_bundle, request.user)
  437. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  438. if request.method == 'POST':
  439. params_form = ParametersFormSet(request.POST)
  440. rerun_form = RerunBundleForm(request.POST, oozie_bundle=oozie_bundle)
  441. if sum([rerun_form.is_valid(), params_form.is_valid()]) == 2:
  442. args = {}
  443. args['deployment_dir'] = app_path
  444. params = {
  445. 'coord-scope': ','.join(rerun_form.cleaned_data['coordinators']),
  446. 'refresh': rerun_form.cleaned_data['refresh'],
  447. 'nocleanup': rerun_form.cleaned_data['nocleanup'],
  448. }
  449. if rerun_form.cleaned_data['start'] and rerun_form.cleaned_data['end']:
  450. date = {
  451. 'date-scope':
  452. '%(start)s::%(end)s' % {
  453. 'start': utc_datetime_format(rerun_form.cleaned_data['start']),
  454. 'end': utc_datetime_format(rerun_form.cleaned_data['end'])
  455. }
  456. }
  457. params.update(date)
  458. properties = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  459. _rerun_bundle(request, job_id, args, params, properties)
  460. request.info(_('Bundle re-running.'))
  461. return redirect(reverse('oozie:list_oozie_bundle', kwargs={'job_id': job_id}))
  462. else:
  463. request.error(_('Invalid submission form: %s' % (rerun_form.errors,)))
  464. return list_oozie_bundle(request, job_id)
  465. else:
  466. rerun_form = RerunBundleForm(oozie_bundle=oozie_bundle)
  467. initial_params = ParameterForm.get_initial_params(oozie_bundle.conf_dict)
  468. params_form = ParametersFormSet(initial=initial_params)
  469. popup = render('dashboard/rerun_bundle_popup.mako', request, {
  470. 'rerun_form': rerun_form,
  471. 'params_form': params_form,
  472. 'action': reverse('oozie:rerun_oozie_bundle', kwargs={'job_id': job_id, 'app_path': app_path}),
  473. }, force_template=True).content
  474. return HttpResponse(json.dumps(popup), mimetype="application/json")
  475. def _rerun_bundle(request, oozie_id, args, params, properties):
  476. try:
  477. submission = Submission(user=request.user, fs=request.fs, jt=request.jt, oozie_id=oozie_id, properties=properties)
  478. job_id = submission.rerun_bundle(params=params, **args)
  479. return job_id
  480. except RestException, ex:
  481. raise PopupException(_("Error re-running bundle %s.") % (oozie_id,),
  482. detail=ex._headers.get('oozie-error-message', ex))
  483. def submit_external_job(request, application_path):
  484. ParametersFormSet = formset_factory(ParameterForm, extra=0)
  485. if request.method == 'POST':
  486. params_form = ParametersFormSet(request.POST)
  487. if params_form.is_valid():
  488. mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
  489. application_name = os.path.basename(application_path)
  490. application_class = Bundle if application_name == 'bundle.xml' else Coordinator if application_name == 'coordinator.xml' else Workflow
  491. mapping[application_class.get_application_path_key()] = application_path
  492. try:
  493. submission = Submission(request.user, fs=request.fs, jt=request.jt, properties=mapping)
  494. job_id = submission.run(application_path)
  495. except RestException, ex:
  496. detail = ex._headers.get('oozie-error-message', ex)
  497. if 'Max retries exceeded with url' in str(detail):
  498. detail = '%s: %s' % (_('The Oozie server is not running'), detail)
  499. LOG.error(smart_str(detail))
  500. raise PopupException(_("Error submitting job %s") % (application_path,), detail=detail)
  501. request.info(_('Oozie job submitted'))
  502. view = 'list_oozie_bundle' if application_name == 'bundle.xml' else 'list_oozie_coordinator' if application_name == 'coordinator.xml' else 'list_oozie_workflow'
  503. return redirect(reverse('oozie:%s' % view, kwargs={'job_id': job_id}))
  504. else:
  505. request.error(_('Invalid submission form: %s' % params_form.errors))
  506. else:
  507. parameters = Submission(request.user, fs=request.fs, jt=request.jt).get_external_parameters(application_path)
  508. initial_params = ParameterForm.get_initial_params(parameters)
  509. params_form = ParametersFormSet(initial=initial_params)
  510. popup = render('editor/submit_job_popup.mako', request, {
  511. 'params_form': params_form,
  512. 'action': reverse('oozie:submit_external_job', kwargs={'application_path': application_path})
  513. }, force_template=True).content
  514. return HttpResponse(json.dumps(popup), mimetype="application/json")
  515. def massaged_workflow_actions_for_json(workflow_actions, oozie_coordinator, oozie_bundle):
  516. actions = []
  517. for action in workflow_actions:
  518. if oozie_coordinator is not None:
  519. setattr(action, 'oozie_coordinator', oozie_coordinator)
  520. if oozie_bundle is not None:
  521. setattr(action, 'oozie_bundle', oozie_bundle)
  522. massaged_action = {
  523. 'id': action.id,
  524. 'log': action.externalId and reverse('jobbrowser.views.job_single_logs', kwargs={'job': action.externalId}) or '',
  525. 'url': action.get_absolute_url(),
  526. 'name': action.name,
  527. 'type': action.type,
  528. 'status': action.status,
  529. 'externalIdUrl': action.externalId and reverse('jobbrowser.views.single_job', kwargs={'job': action.externalId}) or '',
  530. 'externalId': action.externalId and '_'.join(action.externalId.split('_')[-2:]) or '',
  531. 'startTime': format_time(action.startTime),
  532. 'endTime': format_time(action.endTime),
  533. 'retries': action.retries,
  534. 'errorCode': action.errorCode,
  535. 'errorMessage': action.errorMessage,
  536. 'transition': action.transition,
  537. 'data': action.data,
  538. }
  539. actions.append(massaged_action)
  540. return actions
  541. def massaged_coordinator_actions_for_json(coordinator, oozie_bundle):
  542. coordinator_id = coordinator.id
  543. coordinator_actions = coordinator.get_working_actions()
  544. actions = []
  545. related_job_ids = []
  546. if oozie_bundle is not None:
  547. related_job_ids.append('bundle_job_id=%s' %oozie_bundle.id)
  548. for action in coordinator_actions:
  549. related_job_ids.append('coordinator_job_id=%s' % coordinator_id)
  550. massaged_action = {
  551. 'id': action.id,
  552. 'url': action.externalId and reverse('oozie:list_oozie_workflow', kwargs={'job_id': action.externalId}) + '?%s' % '&'.join(related_job_ids) or '',
  553. 'number': action.actionNumber,
  554. 'type': action.type,
  555. 'status': action.status,
  556. 'externalId': action.externalId or '-',
  557. 'externalIdUrl': action.externalId and reverse('oozie:list_oozie_workflow_action', kwargs={'action': action.externalId}) or '',
  558. 'nominalTime': format_time(action.nominalTime),
  559. 'title': action.title,
  560. 'createdTime': format_time(action.createdTime),
  561. 'lastModifiedTime': format_time(action.lastModifiedTime),
  562. 'errorCode': action.errorCode,
  563. 'errorMessage': action.errorMessage,
  564. 'missingDependencies': action.missingDependencies
  565. }
  566. actions.insert(0, massaged_action)
  567. return actions
  568. def massaged_bundle_actions_for_json(bundle):
  569. bundle_actions = bundle.get_working_actions()
  570. actions = []
  571. for action in bundle_actions:
  572. massaged_action = {
  573. 'id': action.coordJobId,
  574. 'url': action.coordJobId and reverse('oozie:list_oozie_coordinator', kwargs={'job_id': action.coordJobId}) + '?bundle_job_id=%s' % bundle.id or '',
  575. 'name': action.coordJobName,
  576. 'type': action.type,
  577. 'status': action.status,
  578. 'externalId': action.coordExternalId or '-',
  579. 'frequency': action.frequency,
  580. 'timeUnit': action.timeUnit,
  581. 'nextMaterializedTime': action.nextMaterializedTime,
  582. 'concurrency': action.concurrency,
  583. 'pauseTime': action.pauseTime,
  584. 'user': action.user,
  585. 'acl': action.acl,
  586. 'timeOut': action.timeOut,
  587. 'coordJobPath': action.coordJobPath,
  588. 'executionPolicy': action.executionPolicy,
  589. 'startTime': action.startTime,
  590. 'endTime': action.endTime,
  591. 'lastAction': action.lastAction
  592. }
  593. actions.insert(0, massaged_action)
  594. return actions
  595. def format_time(st_time):
  596. if st_time is None:
  597. return '-'
  598. elif type(st_time) == time.struct_time:
  599. return time.strftime("%a, %d %b %Y %H:%M:%S", st_time)
  600. else:
  601. return st_time
  602. def massaged_oozie_jobs_for_json(oozie_jobs, user, just_sla=False):
  603. jobs = []
  604. for job in oozie_jobs:
  605. # if job.is_running():
  606. # if job.type == 'Workflow':
  607. # job = get_oozie(user).get_job(job.id)
  608. # elif job.type == 'Coordinator':
  609. # job = get_oozie(user).get_coordinator(job.id)
  610. # else:
  611. # job = get_oozie(user).get_bundle(job.id)
  612. if not just_sla or (just_sla and job.has_sla) and job.appName != 'pig-app-hue-script':
  613. massaged_job = {
  614. 'id': job.id,
  615. 'lastModTime': hasattr(job, 'lastModTime') and job.lastModTime and format_time(job.lastModTime) or None,
  616. 'kickoffTime': hasattr(job, 'kickoffTime') and job.kickoffTime and format_time(job.kickoffTime) or '',
  617. 'nextMaterializedTime': hasattr(job, 'nextMaterializedTime') and job.nextMaterializedTime and format_time(job.nextMaterializedTime) or '',
  618. 'timeOut': hasattr(job, 'timeOut') and job.timeOut or None,
  619. 'endTime': job.endTime and format_time(job.endTime) or None,
  620. 'status': job.status,
  621. 'isRunning': job.is_running(),
  622. 'duration': job.endTime and job.startTime and format_duration_in_millis(( time.mktime(job.endTime) - time.mktime(job.startTime) ) * 1000) or None,
  623. 'appName': job.appName,
  624. 'progress': job.get_progress(),
  625. 'user': job.user,
  626. 'absoluteUrl': job.get_absolute_url(),
  627. 'canEdit': has_job_edition_permission(job, user),
  628. 'killUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'kill'}),
  629. 'suspendUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'suspend'}),
  630. 'resumeUrl': reverse('oozie:manage_oozie_jobs', kwargs={'job_id':job.id, 'action':'resume'}),
  631. 'created': hasattr(job, 'createdTime') and job.createdTime and job.createdTime and ((job.type == 'Bundle' and job.createdTime) or format_time(job.createdTime)),
  632. 'startTime': hasattr(job, 'startTime') and format_time(job.startTime) or None,
  633. 'run': hasattr(job, 'run') and job.run or 0,
  634. 'frequency': hasattr(job, 'frequency') and job.frequency or None,
  635. 'timeUnit': hasattr(job, 'timeUnit') and job.timeUnit or None,
  636. }
  637. jobs.append(massaged_job)
  638. return jobs
  639. def split_oozie_jobs(user, oozie_jobs):
  640. jobs = {}
  641. jobs_running = []
  642. jobs_completed = []
  643. for job in oozie_jobs:
  644. if job.appName != 'pig-app-hue-script':
  645. if job.is_running():
  646. if job.type == 'Workflow':
  647. job = get_oozie(user).get_job(job.id)
  648. elif job.type == 'Coordinator':
  649. job = get_oozie(user).get_coordinator(job.id)
  650. else:
  651. job = get_oozie(user).get_bundle(job.id)
  652. jobs_running.append(job)
  653. else:
  654. jobs_completed.append(job)
  655. jobs['running_jobs'] = sorted(jobs_running, key=lambda w: w.status)
  656. jobs['completed_jobs'] = sorted(jobs_completed, key=lambda w: w.status)
  657. return jobs
  658. def check_job_access_permission(request, job_id):
  659. """
  660. Decorator ensuring that the user has access to the job submitted to Oozie.
  661. Arg: Oozie 'workflow', 'coordinator' or 'bundle' ID.
  662. Return: the Oozie workflow, coordinator or bundle or raise an exception
  663. Notice: its gets an id in input and returns the full object in output (not an id).
  664. """
  665. if job_id is not None:
  666. if job_id.endswith('W'):
  667. get_job = get_oozie(request.user).get_job
  668. elif job_id.endswith('C'):
  669. get_job = get_oozie(request.user).get_coordinator
  670. else:
  671. get_job = get_oozie(request.user).get_bundle
  672. try:
  673. oozie_job = get_job(job_id)
  674. except RestException, ex:
  675. raise PopupException(_("Error accessing Oozie job %s.") % (job_id,),
  676. detail=ex._headers['oozie-error-message'])
  677. if request.user.is_superuser \
  678. or oozie_job.user == request.user.username \
  679. or has_dashboard_jobs_access(request.user):
  680. return oozie_job
  681. else:
  682. message = _("Permission denied. %(username)s does not have the permissions to access job %(id)s.") % \
  683. {'username': request.user.username, 'id': oozie_job.id}
  684. access_warn(request, message)
  685. raise PopupException(message)
  686. def check_job_edition_permission(oozie_job, user):
  687. if has_job_edition_permission(oozie_job, user):
  688. return oozie_job
  689. else:
  690. message = _("Permission denied. %(username)s does not have the permissions to modify job %(id)s.") % \
  691. {'username': user.username, 'id': oozie_job.id}
  692. raise PopupException(message)
  693. def has_job_edition_permission(oozie_job, user):
  694. return user.is_superuser or oozie_job.user == user.username
  695. def has_dashboard_jobs_access(user):
  696. return user.is_superuser or user.has_hue_permission(action="dashboard_jobs_access", app=DJANGO_APPS[0])