views.py 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627
  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 re
  19. import time
  20. import logging
  21. import string
  22. import urlparse
  23. from urllib import quote_plus
  24. from lxml import html
  25. from django.http import HttpResponseRedirect, HttpResponse
  26. from django.utils.functional import wraps
  27. from django.utils.translation import ugettext as _
  28. from django.core.urlresolvers import reverse
  29. from desktop.log.access import access_warn, access_log_level
  30. from desktop.lib.rest.http_client import RestException
  31. from desktop.lib.rest.resource import Resource
  32. from desktop.lib.django_util import render_json, render, copy_query_dict, encode_json_for_js
  33. from desktop.lib.exceptions import MessageException
  34. from desktop.lib.exceptions_renderable import PopupException
  35. from desktop.views import register_status_bar_view
  36. from hadoop import cluster
  37. from hadoop.api.jobtracker.ttypes import ThriftJobPriority, TaskTrackerNotFoundException, ThriftJobState
  38. from hadoop.yarn.clients import get_log_client
  39. from jobbrowser import conf
  40. from jobbrowser.api import get_api, ApplicationNotRunning
  41. from jobbrowser.models import Job, JobLinkage, Tracker, Cluster
  42. import urllib2
  43. def check_job_permission(view_func):
  44. """
  45. Ensure that the user has access to the job.
  46. Assumes that the wrapped function takes a 'jobid' param named 'job'.
  47. """
  48. def decorate(request, *args, **kwargs):
  49. jobid = kwargs['job']
  50. try:
  51. job = get_api(request.user, request.jt).get_job(jobid=jobid)
  52. except ApplicationNotRunning, e:
  53. # reverse() seems broken, using request.path but beware, it discards GET and POST info
  54. return job_not_assigned(request, jobid, request.path)
  55. except Exception, e:
  56. raise PopupException(_('Could not find job %s.') % jobid, detail=e)
  57. if not conf.SHARE_JOBS.get() and not request.user.is_superuser \
  58. and job.user != request.user.username:
  59. raise PopupException(_("You don't have permission to access job %(id)s.") % {'id': jobid})
  60. kwargs['job'] = job
  61. return view_func(request, *args, **kwargs)
  62. return wraps(view_func)(decorate)
  63. def job_not_assigned(request, jobid, path):
  64. if request.GET.get('format') == 'json':
  65. result = {'status': -1, 'message': ''}
  66. try:
  67. get_api(request.user, request.jt).get_job(jobid=jobid)
  68. result['status'] = 0
  69. except ApplicationNotRunning, e:
  70. result['status'] = 1
  71. except Exception, e:
  72. result['message'] = _('Error polling job %s: %s') % (jobid, e)
  73. return HttpResponse(encode_json_for_js(result), mimetype="application/json")
  74. else:
  75. return render('job_not_assigned.mako', request, {'jobid': jobid, 'path': path})
  76. def jobs(request):
  77. user = request.GET.get('user', request.user.username)
  78. state = request.GET.get('state')
  79. text = request.GET.get('text')
  80. retired = request.GET.get('retired')
  81. if request.GET.get('format') == 'json':
  82. jobs = get_api(request.user, request.jt).get_jobs(user=request.user, username=user, state=state, text=text, retired=retired)
  83. json_jobs = [massage_job_for_json(job, request) for job in jobs]
  84. return HttpResponse(encode_json_for_js(json_jobs), mimetype="application/json")
  85. return render('jobs.mako', request, {
  86. 'request': request,
  87. 'state_filter': state,
  88. 'user_filter': user,
  89. 'text_filter': text,
  90. 'retired': retired,
  91. 'filtered': not (state == 'all' and user == '' and text == ''),
  92. 'is_yarn': cluster.is_yarn()
  93. })
  94. def massage_job_for_json(job, request):
  95. job = {
  96. 'id': job.jobId,
  97. 'shortId': job.jobId_short,
  98. 'name': hasattr(job, 'jobName') and job.jobName or '',
  99. 'status': job.status,
  100. 'url': job.jobId and reverse('jobbrowser.views.single_job', kwargs={'job': job.jobId}) or '',
  101. 'logs': job.jobId and reverse('jobbrowser.views.job_single_logs', kwargs={'job': job.jobId}) or '',
  102. 'queueName': hasattr(job, 'queueName') and job.queueName or _('N/A'),
  103. 'priority': hasattr(job, 'priority') and job.priority.lower() or _('N/A'),
  104. 'user': job.user,
  105. 'isRetired': job.is_retired,
  106. 'isMR2': job.is_mr2,
  107. 'mapProgress': hasattr(job, 'mapProgress') and job.mapProgress or '',
  108. 'reduceProgress': hasattr(job, 'reduceProgress') and job.reduceProgress or '',
  109. 'setupProgress': hasattr(job, 'setupProgress') and job.setupProgress or '',
  110. 'cleanupProgress': hasattr(job, 'cleanupProgress') and job.cleanupProgress or '',
  111. 'desiredMaps': job.desiredMaps,
  112. 'desiredReduces': job.desiredReduces,
  113. 'mapsPercentComplete': job.maps_percent_complete,
  114. 'finishedMaps': job.finishedMaps,
  115. 'finishedReduces': job.finishedReduces,
  116. 'reducesPercentComplete': job.reduces_percent_complete,
  117. 'jobFile': hasattr(job, 'jobFile') and job.jobFile or '',
  118. 'launchTimeMs': hasattr(job, 'launchTimeMs') and job.launchTimeMs or '',
  119. 'launchTimeFormatted': hasattr(job, 'launchTimeFormatted') and job.launchTimeFormatted or '',
  120. 'startTimeMs': hasattr(job, 'startTimeMs') and job.startTimeMs or '',
  121. 'startTimeFormatted': hasattr(job, 'startTimeFormatted') and job.startTimeFormatted or '',
  122. 'finishTimeMs': hasattr(job, 'finishTimeMs') and job.finishTimeMs or '',
  123. 'finishTimeFormatted': hasattr(job, 'finishTimeFormatted') and job.finishTimeFormatted or '',
  124. 'durationFormatted': hasattr(job, 'durationFormatted') and job.durationFormatted or '',
  125. 'durationMs': hasattr(job, 'durationInMillis') and job.durationInMillis or '',
  126. 'canKill': (job.status.lower() == 'running' or job.status.lower() == 'pending') and not job.is_mr2 and (request.user.is_superuser or request.user.username == job.user),
  127. 'killUrl': job.jobId and reverse('jobbrowser.views.kill_job', kwargs={'job': job.jobId}) or ''
  128. }
  129. return job
  130. def massage_task_for_json(task):
  131. task = {
  132. 'id': task.taskId,
  133. 'shortId': task.taskId_short,
  134. 'url': task.taskId and reverse('jobbrowser.views.single_task', kwargs={'job': task.jobId, 'taskid': task.taskId}) or '',
  135. 'logs': task.taskAttemptIds and reverse('jobbrowser.views.single_task_attempt_logs', kwargs={'job': task.jobId, 'taskid': task.taskId, 'attemptid': task.taskAttemptIds[-1]}) or '',
  136. 'type': task.taskType
  137. }
  138. return task
  139. @check_job_permission
  140. def single_job(request, job):
  141. def cmp_exec_time(task1, task2):
  142. return cmp(task1.execStartTimeMs, task2.execStartTimeMs)
  143. failed_tasks = job.filter_tasks(task_states=('failed',))
  144. failed_tasks.sort(cmp_exec_time)
  145. recent_tasks = job.filter_tasks(task_states=('running', 'succeeded',))
  146. recent_tasks.sort(cmp_exec_time, reverse=True)
  147. if request.REQUEST.get('format') == 'json':
  148. json_failed_tasks = [massage_task_for_json(task) for task in failed_tasks]
  149. json_recent_tasks = [massage_task_for_json(task) for task in recent_tasks]
  150. json_job = {
  151. 'job': massage_job_for_json(job, request),
  152. 'failedTasks': json_failed_tasks,
  153. 'recentTasks': json_recent_tasks
  154. }
  155. return HttpResponse(encode_json_for_js(json_job), mimetype="application/json")
  156. return render('job.mako', request, {
  157. 'request': request,
  158. 'job': job,
  159. 'failed_tasks': failed_tasks and failed_tasks[:5] or [],
  160. 'recent_tasks': recent_tasks and recent_tasks[:5] or [],
  161. })
  162. @check_job_permission
  163. def job_counters(request, job):
  164. return render("counters.html", request, {"counters": job.counters})
  165. @access_log_level(logging.WARN)
  166. @check_job_permission
  167. def kill_job(request, job):
  168. if request.method != "POST":
  169. raise Exception(_("kill_job may only be invoked with a POST (got a %(method)s).") % dict(method=request.method))
  170. if job.user != request.user.username and not request.user.is_superuser:
  171. access_warn(request, _('Insufficient permission'))
  172. raise MessageException(_("Permission denied. User %(username)s cannot delete user %(user)s's job.") %
  173. dict(username=request.user.username, user=job.user))
  174. job.kill()
  175. cur_time = time.time()
  176. while time.time() - cur_time < 15:
  177. job = Job.from_id(jt=request.jt, jobid=job.jobId)
  178. if job.status not in ["RUNNING", "QUEUED"]:
  179. if request.REQUEST.get("next"):
  180. return HttpResponseRedirect(request.REQUEST.get("next"))
  181. elif request.REQUEST.get("format") == "json":
  182. return HttpResponse(encode_json_for_js({'status': 0}), mimetype="application/json")
  183. else:
  184. raise MessageException("Job Killed")
  185. time.sleep(1)
  186. job = Job.from_id(jt=request.jt, jobid=job.jobId)
  187. raise Exception(_("Job did not appear as killed within 15 seconds."))
  188. @check_job_permission
  189. def job_attempt_logs(request, job, attempt_index=0):
  190. return render("job_attempt_logs.mako", request, {
  191. "attempt_index": attempt_index,
  192. "job": job,
  193. })
  194. @check_job_permission
  195. def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=0):
  196. """For async log retrieval as Yarn servers are very slow"""
  197. try:
  198. attempt_index = int(attempt_index)
  199. attempt = job.job_attempts['jobAttempt'][attempt_index]
  200. log_link = attempt['logsLink']
  201. except (KeyError, RestException), e:
  202. raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
  203. link = '/%s/' % name
  204. params = {}
  205. if offset and int(offset) >= 0:
  206. params['start'] = offset
  207. root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
  208. try:
  209. response = root.get(link, params=params)
  210. log = html.fromstring(response).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
  211. except Exception, e:
  212. log = _('Failed to retrieve log: %s') % e
  213. response = {'log': log}
  214. return HttpResponse(json.dumps(response), mimetype="application/json")
  215. @check_job_permission
  216. def job_single_logs(request, job):
  217. """
  218. Try to smartly detect the most useful task attempt (e.g. Oozie launcher, failed task) and get its MR logs.
  219. """
  220. def cmp_exec_time(task1, task2):
  221. return cmp(task1.execStartTimeMs, task2.execStartTimeMs)
  222. task = None
  223. failed_tasks = job.filter_tasks(task_states=('failed',))
  224. failed_tasks.sort(cmp_exec_time)
  225. if failed_tasks:
  226. task = failed_tasks[0]
  227. else:
  228. task_states = ['running', 'succeeded']
  229. if job.is_mr2:
  230. task_states.append('scheduled')
  231. recent_tasks = job.filter_tasks(task_states=task_states, task_types=('map', 'reduce',))
  232. recent_tasks.sort(cmp_exec_time, reverse=True)
  233. if recent_tasks:
  234. task = recent_tasks[0]
  235. if task is None or not task.taskAttemptIds:
  236. raise PopupException(_("No tasks found for job %(id)s.") % {'id': job.jobId})
  237. return single_task_attempt_logs(request, **{'job': job.jobId, 'taskid': task.taskId, 'attemptid': task.taskAttemptIds[-1]})
  238. @check_job_permission
  239. def tasks(request, job):
  240. """
  241. We get here from /jobs/job/tasks?filterargs, with the options being:
  242. page=<n> - Controls pagination. Defaults to 1.
  243. tasktype=<type> - Type can be one of hadoop.job_tracker.VALID_TASK_TYPES
  244. ("map", "reduce", "job_cleanup", "job_setup")
  245. taskstate=<state> - State can be one of hadoop.job_tracker.VALID_TASK_STATES
  246. ("succeeded", "failed", "running", "pending", "killed")
  247. tasktext=<text> - Where <text> is a string matching info on the task
  248. """
  249. ttypes = request.GET.get('tasktype')
  250. tstates = request.GET.get('taskstate')
  251. ttext = request.GET.get('tasktext')
  252. pagenum = int(request.GET.get('page', 1))
  253. pagenum = pagenum > 0 and pagenum or 1
  254. filters = {
  255. 'task_types': ttypes and set(ttypes.split(',')) or None,
  256. 'task_states': tstates and set(tstates.split(',')) or None,
  257. 'task_text': ttext,
  258. 'pagenum': pagenum,
  259. }
  260. jt = get_api(request.user, request.jt)
  261. task_list = jt.get_tasks(job.jobId, **filters)
  262. page = jt.paginate_task(task_list, pagenum)
  263. filter_params = copy_query_dict(request.GET, ('tasktype', 'taskstate', 'tasktext')).urlencode()
  264. return render("tasks.mako", request, {
  265. 'request': request,
  266. 'filter_params': filter_params,
  267. 'job': job,
  268. 'page': page,
  269. 'tasktype': ttypes,
  270. 'taskstate': tstates,
  271. 'tasktext': ttext
  272. })
  273. @check_job_permission
  274. def single_task(request, job, taskid):
  275. jt = get_api(request.user, request.jt)
  276. job_link = jt.get_job_link(job.jobId)
  277. task = job_link.get_task(taskid)
  278. return render("task.mako", request, {
  279. 'task': task,
  280. 'joblnk': job_link
  281. })
  282. @check_job_permission
  283. def single_task_attempt(request, job, taskid, attemptid):
  284. jt = get_api(request.user, request.jt)
  285. job_link = jt.get_job_link(job.jobId)
  286. task = job_link.get_task(taskid)
  287. try:
  288. attempt = task.get_attempt(attemptid)
  289. except (KeyError, RestException), e:
  290. raise PopupException(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
  291. return render("attempt.mako", request, {
  292. "attempt": attempt,
  293. "taskid": taskid,
  294. "joblnk": job_link,
  295. "task": task
  296. })
  297. @check_job_permission
  298. def single_task_attempt_logs(request, job, taskid, attemptid):
  299. jt = get_api(request.user, request.jt)
  300. job_link = jt.get_job_link(job.jobId)
  301. task = job_link.get_task(taskid)
  302. try:
  303. attempt = task.get_attempt(attemptid)
  304. except (KeyError, RestException), e:
  305. raise KeyError(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
  306. first_log_tab = 0
  307. try:
  308. # Add a diagnostic log
  309. if job_link.is_mr2:
  310. diagnostic_log = attempt.diagnostics
  311. else:
  312. diagnostic_log = ", ".join(task.diagnosticMap[attempt.attemptId])
  313. logs = [diagnostic_log]
  314. # Add remaining logs
  315. logs += [section.strip() for section in attempt.get_task_log()]
  316. log_tab = [i for i, log in enumerate(logs) if log]
  317. if log_tab:
  318. first_log_tab = log_tab[0]
  319. except TaskTrackerNotFoundException:
  320. # Four entries,
  321. # for diagnostic, stdout, stderr and syslog
  322. logs = [_("Failed to retrieve log. TaskTracker not found.")] * 4
  323. except urllib2.URLError:
  324. logs = [_("Failed to retrieve log. TaskTracker not ready.")] * 4
  325. context = {
  326. "attempt": attempt,
  327. "taskid": taskid,
  328. "joblnk": job_link,
  329. "task": task,
  330. "logs": logs,
  331. "first_log_tab": first_log_tab,
  332. }
  333. if request.GET.get('format') == 'python':
  334. return context
  335. elif request.GET.get('format') == 'json':
  336. response = {
  337. "logs": logs,
  338. "isRunning": job.status.lower() in ('running', 'pending', 'prep')
  339. }
  340. return HttpResponse(json.dumps(response), mimetype="application/json")
  341. else:
  342. return render("attempt_logs.mako", request, context)
  343. @check_job_permission
  344. def task_attempt_counters(request, job, taskid, attemptid):
  345. """
  346. We get here from /jobs/jobid/tasks/taskid/attempts/attemptid/counters
  347. (phew!)
  348. """
  349. job_link = JobLinkage(request.jt, job.jobId)
  350. task = job_link.get_task(taskid)
  351. attempt = task.get_attempt(attemptid)
  352. counters = {}
  353. if attempt:
  354. counters = attempt.counters
  355. return render("counters.html", request, {'counters':counters})
  356. @access_log_level(logging.WARN)
  357. def kill_task_attempt(request, attemptid):
  358. """
  359. We get here from /jobs/jobid/tasks/taskid/attempts/attemptid/kill
  360. TODO: security
  361. """
  362. ret = request.jt.kill_task_attempt(request.jt.thriftattemptid_from_string(attemptid))
  363. return render_json({})
  364. def trackers(request):
  365. """
  366. We get here from /trackers
  367. """
  368. trackers = get_tasktrackers(request)
  369. return render("tasktrackers.mako", request, {'trackers':trackers})
  370. def single_tracker(request, trackerid):
  371. jt = get_api(request.user, request.jt)
  372. try:
  373. tracker = jt.get_tracker(trackerid)
  374. except Exception, e:
  375. raise PopupException(_('The tracker could not be contacted.'), detail=e)
  376. return render("tasktracker.mako", request, {'tracker':tracker})
  377. def container(request, node_manager_http_address, containerid):
  378. jt = get_api(request.user, request.jt)
  379. try:
  380. tracker = jt.get_tracker(node_manager_http_address, containerid)
  381. except Exception, e:
  382. # TODO: add a redirect of some kind
  383. raise PopupException(_('The container disappears as soon as the job finishes.'), detail=e)
  384. return render("container.mako", request, {'tracker':tracker})
  385. def clusterstatus(request):
  386. """
  387. We get here from /clusterstatus
  388. """
  389. return render("clusterstatus.html", request, Cluster(request.jt))
  390. def queues(request):
  391. """
  392. We get here from /queues
  393. """
  394. return render("queues.html", request, { "queuelist" : request.jt.queues()})
  395. @check_job_permission
  396. def set_job_priority(request, job):
  397. """
  398. We get here from /jobs/job/setpriority?priority=PRIORITY
  399. """
  400. priority = request.GET.get("priority")
  401. jid = request.jt.thriftjobid_from_string(job.jobId)
  402. request.jt.set_job_priority(jid, ThriftJobPriority._NAMES_TO_VALUES[priority])
  403. return render_json({})
  404. CONF_VARIABLE_REGEX = r"\$\{(.+)\}"
  405. def make_substitutions(conf):
  406. """
  407. Substitute occurences of ${foo} with conf[foo], recursively, in all the values
  408. of the conf dict.
  409. Note that the Java code may also substitute Java properties in, which
  410. this code does not have.
  411. """
  412. r = re.compile(CONF_VARIABLE_REGEX)
  413. def sub(s, depth=0):
  414. # Malformed / malicious confs could make this loop infinitely
  415. if depth > 100:
  416. logging.warn("Max recursion depth exceeded when substituting jobconf value: %s" % s)
  417. return s
  418. m = r.search(s)
  419. if m:
  420. for g in [g for g in m.groups() if g in conf]:
  421. substr = "${%s}" % g
  422. s = s.replace(substr, sub(conf[g], depth+1))
  423. return s
  424. for k, v in conf.items():
  425. conf[k] = sub(v)
  426. return conf
  427. ##################################
  428. ## Helper functions
  429. def get_shorter_id(hadoop_job_id):
  430. return "_".join(hadoop_job_id.split("_")[-2:])
  431. def format_counter_name(s):
  432. """
  433. Makes counter/config names human readable:
  434. FOOBAR_BAZ -> "Foobar Baz"
  435. foo_barBaz -> "Foo Bar Baz"
  436. """
  437. def splitCamels(s):
  438. """ Convert "fooBar" to "foo bar" """
  439. return re.sub(r'[a-z][A-Z]',
  440. lambda x: x.group(0)[0] + " " + x.group(0)[1].lower(),
  441. s)
  442. return string.capwords(re.sub('_', ' ', splitCamels(s)).lower())
  443. def get_state_link(request, option=None, val='', VALID_OPTIONS = ("state", "user", "text", "taskstate")):
  444. """
  445. constructs the query string for the state of the current query for the jobs page.
  446. pass in the request, and an optional option/value pair; these are used for creating
  447. links to turn on the filter, while preserving the other present settings.
  448. """
  449. states = []
  450. val = quote_plus(val)
  451. assert option is None or option in VALID_OPTIONS
  452. states = dict()
  453. for o in VALID_OPTIONS:
  454. if o in request.GET:
  455. states[o] = request.GET[o]
  456. if option is not None:
  457. states[option] = val
  458. return "&".join([ "%s=%s" % (key, quote_plus(value)) for key, value in states.iteritems() ])
  459. ## All Unused below
  460. # DEAD?
  461. def dock_jobs(request):
  462. username = request.user.username
  463. matching_jobs = get_job_count_by_state(request, username)
  464. return render("jobs_dock_info.mako", request, {
  465. 'jobs': matching_jobs
  466. }, force_template=True)
  467. register_status_bar_view(dock_jobs)
  468. def get_tasktrackers(request):
  469. """
  470. Return a ThriftTaskTrackerStatusList object containing all task trackers
  471. """
  472. return [ Tracker(tracker) for tracker in request.jt.all_task_trackers().trackers]
  473. def get_single_job(request, jobid):
  474. """
  475. Returns the job which matches jobid.
  476. """
  477. return Job.from_id(jt=request.jt, jobid=jobid)
  478. def get_job_count_by_state(request, username):
  479. """
  480. Returns the number of comlpeted, running, and failed jobs for a user.
  481. """
  482. res = {
  483. 'completed': 0,
  484. 'running': 0,
  485. 'failed': 0,
  486. 'killed': 0,
  487. 'all': 0
  488. }
  489. jobcounts = request.jt.get_job_count_by_user(username)
  490. res['completed'] = jobcounts.nSucceeded
  491. res['running'] = jobcounts.nPrep + jobcounts.nRunning
  492. res['failed'] = jobcounts.nFailed
  493. res['killed'] = jobcounts.nKilled
  494. res['all'] = res['completed'] + res['running'] + res['failed'] + res['killed']
  495. return res
  496. def jobbrowser(request):
  497. """
  498. jobbrowser.jsp - a - like.
  499. """
  500. # TODO(bc): Is this view even reachable?
  501. def check_job_state(state):
  502. return lambda job: job.status == state
  503. status = request.jt.cluster_status()
  504. alljobs = [] #get_matching_jobs(request)
  505. runningjobs = filter(check_job_state('RUNNING'), alljobs)
  506. completedjobs = filter(check_job_state('COMPLETED'), alljobs)
  507. failedjobs = filter(check_job_state('FAILED'), alljobs)
  508. killedjobs = filter(check_job_state('KILLED'), alljobs)
  509. jobqueues = request.jt.queues()
  510. return render("jobbrowser.html", request, {
  511. "clusterstatus" : status,
  512. "queues" : jobqueues,
  513. "alljobs" : alljobs,
  514. "runningjobs" : runningjobs,
  515. "failedjobs" : failedjobs,
  516. "killedjobs" : killedjobs,
  517. "completedjobs" : completedjobs
  518. })