Browse Source

HUE-697 [jobbrowser] Ability to browse MR2 jobs

Jobbrowser can now lists jobs and their details by using the YARN apis.
Currently, the Resource Manager API is used for listing the jobs and then
the MapReduce API is used when the job is running or otherwise the History
Server API is used. The current convention is to use application_ID when
the job is running and then job_ID when it is finished.

Compatible and tested with MR1 and MR2.
Tests added with YARN API mocked.
3 YARN API implemented and available in libs.
Retrieves logs with a curl as YARN API does not support it.
Asynchronously fetch the logs as it is slow.
With YARN, support a bew job_attempt page.
yarn_models.py created.
YARN API are all read only.
job URL updated in jobsub, oozie, beeswax.
Beeswax now displaying logs and the MR jobs ids.
Rename 'jobid' to 'job' in all the urls and make the check_job_permission
return the job object.
Added 3 new URL Hadoop YARN ini properties
Romain Rigaux 13 years ago
parent
commit
36ffc42
33 changed files with 1479 additions and 376 deletions
  1. 1 2
      apps/beeswax/src/beeswax/templates/watch_results.mako
  2. 1 2
      apps/beeswax/src/beeswax/templates/watch_wait.mako
  3. 2 3
      apps/beeswax/src/beeswax/tests.py
  4. 12 6
      apps/beeswax/src/beeswax/views.py
  5. 218 0
      apps/jobbrowser/src/jobbrowser/api.py
  6. 14 10
      apps/jobbrowser/src/jobbrowser/models.py
  7. 10 6
      apps/jobbrowser/src/jobbrowser/templates/attempt.mako
  8. 4 4
      apps/jobbrowser/src/jobbrowser/templates/attempt_logs.mako
  9. 1 1
      apps/jobbrowser/src/jobbrowser/templates/clusterstatus.html
  10. 70 13
      apps/jobbrowser/src/jobbrowser/templates/job.mako
  11. 113 0
      apps/jobbrowser/src/jobbrowser/templates/job_attempt_logs.mako
  12. 1 1
      apps/jobbrowser/src/jobbrowser/templates/jobbrowser.html
  13. 84 5
      apps/jobbrowser/src/jobbrowser/templates/jobbrowser_components.mako
  14. 30 27
      apps/jobbrowser/src/jobbrowser/templates/jobs.mako
  15. 9 5
      apps/jobbrowser/src/jobbrowser/templates/task.mako
  16. 13 13
      apps/jobbrowser/src/jobbrowser/templates/tasks.mako
  17. 206 1
      apps/jobbrowser/src/jobbrowser/tests.py
  18. 16 13
      apps/jobbrowser/src/jobbrowser/urls.py
  19. 154 257
      apps/jobbrowser/src/jobbrowser/views.py
  20. 206 0
      apps/jobbrowser/src/jobbrowser/yarn_models.py
  21. 1 1
      apps/jobsub/src/jobsub/templates/workflow.mako
  22. 2 2
      apps/oozie/src/oozie/templates/dashboard/list_oozie_workflow.mako
  23. 1 1
      apps/oozie/src/oozie/templates/dashboard/list_oozie_workflow_action.mako
  24. 1 1
      apps/oozie/src/oozie/templates/editor/gen/workflow-graph-status.xml.mako
  25. 9 0
      desktop/conf.dist/hue.ini
  26. 9 0
      desktop/conf/pseudo-distributed.ini.tmpl
  27. 2 2
      desktop/core/src/desktop/lib/rest/resource.py
  28. 3 0
      desktop/libs/hadoop/src/hadoop/cluster.py
  29. 11 0
      desktop/libs/hadoop/src/hadoop/conf.py
  30. 16 0
      desktop/libs/hadoop/src/hadoop/yarn/__init__.py
  31. 93 0
      desktop/libs/hadoop/src/hadoop/yarn/history_server_api.py
  32. 97 0
      desktop/libs/hadoop/src/hadoop/yarn/mapreduce_api.py
  33. 69 0
      desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

+ 1 - 2
apps/beeswax/src/beeswax/templates/watch_results.mako

@@ -80,9 +80,8 @@ ${layout.menubar(section='query')}
 			        %>
 				 	% if n_jobs > 0:
 						<li class="nav-header">${mr_jobs} (${n_jobs})</li>
-
 						% for jobid in hadoop_jobs:
-						<li><a href="${url("jobbrowser.views.single_job", jobid=jobid)}">${jobid.replace("job_", "")}</a></li>
+						    <li><a href="${url("jobbrowser.views.single_job", job=jobid.replace('application', 'job'))}">${ jobid.replace("application_", "") }</a></li>
 						% endfor
 					% else:
 						<li class="nav-header">${mr_jobs}</li>

+ 1 - 2
apps/beeswax/src/beeswax/templates/watch_wait.mako

@@ -39,9 +39,8 @@ ${layout.menubar(section='query')}
 			        %>
 				 	% if n_jobs > 0:
 						<li id="jobsHeader" class="nav-header">${mr_jobs} (${n_jobs})</li>
-
 						% for jobid in hadoop_jobs:
-						<li><a class="jobLink" href="${url("jobbrowser.views.single_job", jobid=jobid)}">${jobid.replace("job_", "")}</a></li>
+						    <li><a class="jobLink" href="${url("jobbrowser.views.single_job", job=jobid)}">${jobid.replace("job_", "")}</a></li>
 						% endfor
 					% else:
 						<li id="jobsHeader" class="nav-header">${mr_jobs}</li>

+ 2 - 3
apps/beeswax/src/beeswax/tests.py

@@ -15,8 +15,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#
-# Tests for beeswax
 
 import cStringIO
 import gzip
@@ -1193,8 +1191,9 @@ Starting Job = job_201003191517_0002, Tracking URL = http://localhost:50030/jobd
     --- we should be ignoring duplicates ---
 Starting Job = job_201003191517_0002, Tracking URL = http://localhost:50030/jobdetails.jsp?jobid=job_201003191517_0002
 Starting Job = job_201003191517_0003, Tracking URL = http://localhost:50030/jobdetails.jsp?jobid=job_201003191517_0003
+12/12/27 10:48:22 INFO mapreduce.Job: The url to track the job: http://localhost:8088/proxy/application_1356251510842_0022/
 """
-  assert_equal(["job_201003191517_0002", "job_201003191517_0003"],
+  assert_equal(["job_201003191517_0002", "job_201003191517_0003", "application_1356251510842_0022"],
     beeswax.views._parse_out_hadoop_jobs(sample_log))
   assert_equal([], beeswax.views._parse_out_hadoop_jobs("nothing to see here"))
 

+ 12 - 6
apps/beeswax/src/beeswax/views.py

@@ -578,9 +578,7 @@ def watch_query_refresh_json(request, id):
   log = dbms.get(request.user, query_history.get_query_server_config()).get_log(handle)
 
   jobs = _parse_out_hadoop_jobs(log)
-  job_urls = {}
-  for job in jobs:
-    job_urls[job] = reverse('jobbrowser.views.single_job', kwargs=dict(jobid=job))
+  job_urls = dict([(job, reverse('jobbrowser.views.single_job', kwargs=dict(job=job))) for job in jobs])
 
   result = {
     'log': log,
@@ -1244,6 +1242,8 @@ def _parse_query_context(context):
 
 
 HADOOP_JOBS_RE = re.compile("(http[^\s]*/jobdetails.jsp\?jobid=([a-z0-9_]*))")
+HADOOP_YARN_JOBS_RE = re.compile("(http[^\s]*/proxy/([a-z0-9_]+?)/)")
+
 def _parse_out_hadoop_jobs(log):
   """
   Ideally, Hive would tell us what jobs it has run directly
@@ -1251,6 +1251,7 @@ def _parse_out_hadoop_jobs(log):
   to look for URLs to those jobs.
   """
   ret = []
+
   for match in HADOOP_JOBS_RE.finditer(log):
     full_job_url, job_id = match.groups()
     # We ignore full_job_url for now, but it may
@@ -1258,9 +1259,14 @@ def _parse_out_hadoop_jobs(log):
     # correctly.
 
     # Ignore duplicates
-    if job_id in ret:
-      continue
-    ret.append(job_id)
+    if job_id not in ret:
+      ret.append(job_id)
+
+  for match in HADOOP_YARN_JOBS_RE.finditer(log):
+    full_job_url, job_id = match.groups()
+    if job_id not in ret:
+      ret.append(job_id)
+
   return ret
 
 

+ 218 - 0
apps/jobbrowser/src/jobbrowser/api.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from desktop.lib.paginator import Paginator
+
+from hadoop import cluster
+from hadoop.api.jobtracker.ttypes import ThriftJobPriority, TaskTrackerNotFoundException, ThriftJobState
+
+import hadoop.yarn.history_server_api as history_server_api
+import hadoop.yarn.mapreduce_api as mapreduce_api
+import hadoop.yarn.resource_manager_api as resource_manager_api
+
+from jobbrowser.conf import SHARE_JOBS
+from jobbrowser.models import Job, JobLinkage, TaskList
+from jobbrowser.yarn_models import Application, Job as YarnJob
+
+
+LOG = logging.getLogger(__name__)
+
+_DEFAULT_OBJ_PER_PAGINATION = 10
+
+
+def get_api(user, jt):
+  if cluster.is_yarn():
+    return YarnApi(user)
+  else:
+    return JtApi(jt)
+
+
+class JobBrowserApi(object):
+
+  def paginate_task(self, task_list, pagenum):
+    paginator = Paginator(task_list, _DEFAULT_OBJ_PER_PAGINATION)
+    return paginator.page(pagenum)
+
+
+class JtApi(JobBrowserApi):
+  def __init__(self, jt):
+    self.jt = jt
+
+  def get_job_link(self, jobid):
+    return JobLinkage(self.jt, jobid)
+
+  def get_job(self, jobid):
+    return Job.from_id(jt=self.jt, jobid=jobid)
+
+  def get_jobs(self, user, **kwargs):
+    """
+    Returns an array of jobs where the returned
+    jobs are matched by the provided filter arguments.
+
+    If a filter argument is in kwargs it will supersede the same argument
+    in the request object.
+
+    Filter arguments may be jobid, pools, user, tasks, text and state.
+
+    Filter by user ownership if check_permission is set to true.
+    """
+    jobfunc = {
+       "completed" : (self.jt.completed_jobs, ThriftJobState.SUCCEEDED),
+       # Succeeded and completed are synonyms here.
+       "succeeded" : (self.jt.completed_jobs, ThriftJobState.SUCCEEDED),
+       "running" : (self.jt.running_jobs, ThriftJobState.RUNNING),
+       "failed" : (self.jt.failed_jobs, ThriftJobState.FAILED),
+       "killed" : (self.jt.killed_jobs, ThriftJobState.KILLED),
+       "all" : (self.jt.all_jobs, None),
+       None : (self.jt.all_jobs, None)
+    }
+
+    selection = kwargs.pop('state')
+    retired = kwargs.pop('retired')
+
+    jobs = jobfunc[selection][0]().jobs
+
+    if retired:
+      jobs += self.jt.retired_jobs(jobfunc[selection][1]).jobs
+
+    return self.filter_jobs(user, jobs, **kwargs)
+
+  def filter_jobs(self, user, jobs, **kwargs):
+    check_permission = not SHARE_JOBS.get() and not user.is_superuser
+
+    return [Job.from_thriftjob(self.jt, j)
+            for j in self._filter_jobs(jobs, **kwargs)
+            if not check_permission or user.is_superuser or j.profile.user == user.username]
+
+  def _filter_jobs(self, jobs, username=None, text=None):
+    def predicate(job):
+      """
+      Return True if a ThriftJobInProgress structure matches the supplied filters.
+
+      If a filter argument is None, everything matches it.
+      """
+      if username and username not in job.profile.user:
+        return False
+
+      if text:
+        search = text.lower()
+        # These fields are chosen to match those displayed by the JT UI
+        saw_text = False
+        for t in [job.profile.user,
+                  job.profile.name,
+                  job.jobID.asString,
+                  job.profile.queueName,
+                  job.priorityAsString
+                  ]:
+          if search in t.lower():
+            saw_text = True
+            break
+
+        if not saw_text:
+          return False
+
+      return True
+
+    return filter(predicate, jobs)
+
+  def get_tasks(self, jobid, **filters):
+    return TaskList.select(self.jt,
+                           jobid,
+                           filters['task_types'],
+                           filters['task_states'],
+                           filters['task_text'],
+                           _DEFAULT_OBJ_PER_PAGINATION,
+                           _DEFAULT_OBJ_PER_PAGINATION * (filters['pagenum'] - 1))
+
+
+class YarnApi(JobBrowserApi):
+  """
+  List all the jobs with Resource Manager API.
+  Get running single job information with MapReduce API.
+  Get finished single job information with History Server API.
+
+  The trick is that we use appid when the job is running and jobid when it is finished.
+  We also suppose that each app id has only one MR job id.
+  e.g. job_1355791146953_0105, application_1355791146953_0105
+
+  A better alternative might be to call the Resource Manager instead of relying on the type of job id.
+  The perfect solution would be to have all this logic embedded
+  """
+  def __init__(self, user):
+    self.user = user
+    self.resource_manager_api = resource_manager_api.get_resource_manager()
+    self.mapreduce_api = mapreduce_api.get_mapreduce_api()
+    self.history_server_api = history_server_api.get_history_server_api()
+
+  def get_job_link(self, job_id):
+    return self.get_job(job_id)
+
+  def get_jobs(self, user, **kwargs):
+    state_filters = {'running': 'UNDEFINED', 'completed': 'SUCCEEDED', 'failed': 'FAILED', 'killed': 'KILLED', }
+    filters = {}
+
+    if kwargs['username']:
+      filters['user'] = kwargs['username']
+    if kwargs['state'] and kwargs['state'] != 'all':
+      filters['finalStatus'] = state_filters[kwargs['state']]
+
+    json = self.resource_manager_api.apps(**filters)
+    if json['apps']:
+      jobs = [Application(app) for app in json['apps']['app']]
+    else:
+      return []
+
+    if kwargs['text']:
+      text = kwargs['text'].lower()
+      jobs = filter(lambda job:
+                    text in job.name.lower() or
+                    text in job.id.lower() or
+                    text in job.user.lower() or
+                    text in job.queue.lower(), jobs)
+
+    return self.filter_jobs(user, jobs)
+
+  def filter_jobs(self, user, jobs, **kwargs):
+    check_permission = not SHARE_JOBS.get() and not user.is_superuser
+
+    return filter(lambda job:
+                  not check_permission or
+                  user.is_superuser or
+                  job.user == user.username, jobs)
+
+  def get_job(self, jobid):
+    """
+    Try first as if it was a running job, then as a finished job.
+    """
+    try:
+      if jobid.startswith('application'):
+        json = self.mapreduce_api.job(self.user, jobid.replace('application', 'job'))
+        return YarnJob(self.mapreduce_api, json['job'])
+    except Exception, e:
+      LOG.info('Job %s not running: %s' % (jobid, e))
+    jobid = jobid.replace('application', 'job')
+    json = self.history_server_api.job(self.user, jobid)
+    return YarnJob(self.history_server_api, json['job'])
+
+  def get_tasks(self, jobid, **filters):
+    filters.pop('pagenum')
+    return self.get_job(jobid).filter_tasks(**filters)
+
+  def get_task(self, jobid, task_id):
+    return self.get_job(jobid).task(task_id)

+ 14 - 10
apps/jobbrowser/src/jobbrowser/models.py

@@ -15,28 +15,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from desktop.lib.view_util import format_duration_in_millis
-from desktop.lib import i18n
-from hadoop import job_tracker
-from hadoop import confparse
-from urlparse import urlparse, urlunparse
-
 import datetime
 import logging
 import lxml.html
 import re
 import urllib2
 
-from django.utils.translation import ugettext as _
+from urlparse import urlparse, urlunparse
 
-import hadoop.api.jobtracker.ttypes as ttypes
+from desktop.lib.view_util import format_duration_in_millis
+from desktop.lib import i18n
+from hadoop import job_tracker
+from hadoop import confparse
 from hadoop.api.jobtracker.ttypes import JobNotFoundException
 
-from django.utils.translation import ugettext as _
+import hadoop.api.jobtracker.ttypes as ttypes
 from desktop.lib.exceptions_renderable import PopupException
 
+from django.utils.translation import ugettext as _
+
+
 LOGGER = logging.getLogger(__name__)
 
+
 class JobLinkage(object):
   """
   A thin representation of a job, without much of the details.
@@ -52,6 +53,7 @@ class JobLinkage(object):
     self._jobtracker = jobtracker
     self.jobId = jobid
     self.jobId_short = "_".join(jobid.split("_")[-2:])
+    self.is_mr2 = False
 
   def get_task(self, task_id):
     """Retrieve a TaskInProgress from hadoop."""
@@ -73,7 +75,7 @@ class Job(JobLinkage):
     return getattr(self, item)
 
   @staticmethod
-  def from_id(jt, jobid):
+  def from_id(jt, jobid, is_finished=False):
     """
       Returns a Job instance given a job tracker interface and an id. The job tracker interface is typically
       located in request.jt.
@@ -114,6 +116,7 @@ class Job(JobLinkage):
     self._full_job_conf = None
     self._init_attributes()
     self.is_retired = hasattr(thriftJob, 'is_retired')
+    self.is_mr2 = False
 
   @property
   def counters(self):
@@ -364,6 +367,7 @@ class Task(object):
     self.counters = self.task.counters
     self.failed = self.task.failed
     self.complete = self.task.complete
+    self.is_mr2 = False
 
   def get_attempt(self, id):
     """

+ 10 - 6
apps/jobbrowser/src/jobbrowser/templates/attempt.mako

@@ -29,10 +29,10 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
                     <li class="nav-header">${_('Attempt ID')}</li>
                     <li>${attempt.attemptId_short}</li>
                     <li class="nav-header">${_('Task')}</li>
-                    <li><a href="${url('jobbrowser.views.single_task', jobid=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a>
+                    <li><a href="${url('jobbrowser.views.single_task', job=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a>
                     </li>
                     <li class="nav-header">${_('Job')}</li>
-                    <li><a href="${url('jobbrowser.views.single_job', jobid=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
+                    <li><a href="${url('jobbrowser.views.single_job', job=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
                     <li class="nav-header">${_('Status')}</li>
                     <li>
                         <%
@@ -54,7 +54,7 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
             <ul id="tabs" class="nav nav-tabs">
                 <li class="active"><a href="#metadata" data-toggle="tab">${_('Metadata')}</a></li>
                 <li><a href="#counters" data-toggle="tab">${_('Counters')}</a></li>
-                <li><a href="${ url('jobbrowser.views.single_task_attempt_logs', jobid=task.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }">${_('Logs')}</a></li>
+                <li><a href="${ url('jobbrowser.views.single_task_attempt_logs', job=task.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }">${_('Logs')}</a></li>
             </ul>
 
             <div class="tab-content">
@@ -73,7 +73,7 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
                         </tr>
                         <tr>
                             <td>${_('Task ID')}</td>
-                            <td><a href="${url('jobbrowser.views.single_task', jobid=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a></td>
+                            <td><a href="${url('jobbrowser.views.single_task', job=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a></td>
                         </tr>
                         <tr>
                             <td>${_('Task Type')}</td>
@@ -81,7 +81,7 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
                         </tr>
                         <tr>
                             <td>${_('JobId')}</td>
-                            <td><a href="${url('jobbrowser.views.single_job', jobid=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></td>
+                            <td><a href="${url('jobbrowser.views.single_job', job=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></td>
                         </tr>
                         <tr>
                             <td>${_('State')}</td>
@@ -116,7 +116,11 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
                 </div>
 
                 <div class="tab-pane" id="counters">
-                    ${comps.task_counters(task.counters)}
+                    % if task.is_mr2:
+                        ${ comps.task_counters_mr2(task.counters) }
+                    % else:
+                        ${ comps.task_counters(task.counters) }
+                    % endif
                 </div>
             </div>
         </div>

+ 4 - 4
apps/jobbrowser/src/jobbrowser/templates/attempt_logs.mako

@@ -29,10 +29,10 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
                     <li class="nav-header">${_('Attempt ID')}</li>
                     <li>${attempt.attemptId_short}</li>
                     <li class="nav-header">${_('Task')}</li>
-                    <li><a href="${url('jobbrowser.views.single_task', jobid=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a>
+                    <li><a href="${url('jobbrowser.views.single_task', job=joblnk.jobId, taskid=taskid)}" title="${_('View this task')}">${task.taskId_short}</a>
                     </li>
                     <li class="nav-header">${_('Job')}</li>
-                    <li><a href="${url('jobbrowser.views.single_job', jobid=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
+                    <li><a href="${url('jobbrowser.views.single_job', job=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
                     <li class="nav-header">${_('Status')}</li>
                     <li>
                         <%
@@ -52,8 +52,8 @@ ${commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=a
 
         <div class="span10">
             <ul class="nav nav-tabs">
-                <li><a href="${ url('jobbrowser.views.single_task_attempt', jobid=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }#tmetadata">${_('Metadata')}</a></li>
-                <li><a href="${ url('jobbrowser.views.single_task_attempt', jobid=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }#tcounters">${_('Counters')}</a></li>
+                <li><a href="${ url('jobbrowser.views.single_task_attempt', job=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }#tmetadata">${_('Metadata')}</a></li>
+                <li><a href="${ url('jobbrowser.views.single_task_attempt', job=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }#tcounters">${_('Counters')}</a></li>
                 <li class="active"><a href="#logs" data-toggle="tab">${_('Logs')}</a></li>
             </ul>
 

+ 1 - 1
apps/jobbrowser/src/jobbrowser/templates/clusterstatus.html

@@ -48,7 +48,7 @@ limitations under the License.
     <td>{{ clusterstatus.state }}
   </tr>
     </th>
-  
+
 </table>
 
 

+ 70 - 13
apps/jobbrowser/src/jobbrowser/templates/job.mako

@@ -19,7 +19,7 @@
 <%!
   import os
   from jobbrowser.views import format_counter_name
-  from filebrowser.views import location_to_url  
+  from filebrowser.views import location_to_url
   from desktop.views import commonheader, commonfooter
 
   from django.template.defaultfilters import urlencode
@@ -40,10 +40,16 @@
             <tr>
                 <td data-row-selector-exclude="true">
                 %if task.taskAttemptIds:
-                    <a href="${ url('jobbrowser.views.single_task_attempt_logs', jobid=task.jobId, taskid=task.taskId, attemptid=task.taskAttemptIds[-1]) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
+                    <a href="${ url('jobbrowser.views.single_task_attempt_logs', job=task.jobId, taskid=task.taskId, attemptid=task.taskAttemptIds[-1]) }"
+                        data-row-selector="true"><i class="icon-tasks"></i>
+                    </a>
                 %endif
                 </td>
-                <td><a title="${_('View this task')}" href="${ url('jobbrowser.views.single_task', jobid=job.jobId, taskid=task.taskId) }" data-row-selector="true">${task.taskId_short}</a></td>
+                <td>
+                    <a title="${_('View this task')}" href="${ url('jobbrowser.views.single_task', job=job.jobId, taskid=task.taskId) }"
+                        data-row-selector-exclude="true">${task.taskId_short}
+                    </a>
+                </td>
                 <td>${task.taskType}</td>
             </tr>
             % endfor
@@ -83,6 +89,7 @@
         </tr>
     % endfor
 </%def>
+
 ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short), "jobbrowser", user)}
 
 <div class="container-fluid">
@@ -100,12 +107,12 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                         ${comps.get_status(job)}
                     </li>
                     <li class="nav-header">${_('Logs')}</li>
-                    <li><a href="${ url('jobbrowser.views.job_single_logs', jobid=job.jobId) }">${_('View logs')}</a></li>
+                    <li><a href="${ url('jobbrowser.views.job_single_logs', job=job.jobId) }">${_('View logs')}</a></li>
                     % if job.status.lower() == 'running' or job.status.lower() == 'pending':
                         <li class="nav-header">${_('Kill Job')}</li>
                         <li>
                           <a href="#" title="${_('Kill this job')}" onclick="$('#kill-job').submit()">${_('Kill this job')}</a>
-                          <form id="kill-job" action="${url('jobbrowser.views.kill_job', jobid=job.jobId)}?next=${request.get_full_path()|urlencode}" method="POST"></form>
+                          <form id="kill-job" action="${url('jobbrowser.views.kill_job', job=job.jobId)}?next=${request.get_full_path()|urlencode}" method="POST"></form>
                         </li>
                     % endif
                     % if not job.is_retired:
@@ -142,16 +149,48 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
         </div>
         <div class="span10">
             <ul class="nav nav-tabs">
-                <li class="active"><a href="#tasks" data-toggle="tab">${_('Tasks')}</a></li>
+                % if job.is_mr2:
+                    <li class="active"><a href="#attempts" data-toggle="tab">${_('Attempts')}</a></li>
+                    <li><a href="#tasks" data-toggle="tab">${_('Tasks')}</a></li>
+                % else:
+                    <li class="active"><a href="#tasks" data-toggle="tab">${_('Tasks')}</a></li>
+                % endif
                 <li><a href="#metadata" data-toggle="tab">${_('Metadata')}</a></li>
                 % if not job.is_retired:
-                <li><a href="#counters" data-toggle="tab">${_('Counters')}</a></li>
+                    <li><a href="#counters" data-toggle="tab">${_('Counters')}</a></li>
                 % endif
             </ul>
 
             <div class="tab-content">
-                <div class="tab-pane active" id="tasks">
-                    % if job.is_retired:
+                % if job.is_mr2:
+                    <div class="tab-pane active" id="attempts">
+                        <table id="jobAttemptTable" class="table table-striped table-condensed">
+                            <thead>
+                                <th>${_('Logs')}</th>
+                                <th>${_('Id')}</th>
+                                <th>${_('Container')}</th>
+                            </thead>
+                            <tbody>
+                                % for attempt in job.job_attempts['jobAttempt']:
+                                    <tr>
+                                        <td>
+                                            <a href="${ url('jobbrowser.views.job_attempt_logs', job=job.jobId, attempt_index=loop.index) }"
+                                                data-row-selector="true">
+                                                <i class="icon-tasks"></i>
+                                            </a>
+                                        </td>
+                                        <td>${ attempt['id'] }</td>
+                                        <td>${ attempt['containerId'] }</td>
+                                    </tr>
+                                % endfor
+                            </tbody>
+                        </table>
+                    </div>
+                    <div class="tab-pane" id="tasks">
+                % else:
+                    <div class="tab-pane active" id="tasks">
+                % endif
+                    % if job.is_retired and not job.is_mr2:
                        ${ _('This jobs is ')} <span class="label label-warning">${ _('retired') }</span> ${ _(' and so has little information available.') }
                        <br/>
                        <br/>
@@ -159,7 +198,7 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                         %if failed_tasks:
                             <div>
                                 <h3>
-                                    <a href="${url('jobbrowser.views.tasks', jobid=job.jobId)}?taskstate=failed">${_('View Failed Tasks')} &raquo;</a>
+                                    <a href="${url('jobbrowser.views.tasks', job=job.jobId)}?taskstate=failed">${_('View Failed Tasks')} &raquo;</a>
                                     ${_('Failed Tasks')}
                                 </h3>
                                 <div>
@@ -168,7 +207,7 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                             </div>
                         %endif
                         <div>
-                            <a style="float:right;margin-right:10px" href="${url('jobbrowser.views.tasks', jobid=job.jobId)}">${_('View All Tasks')} &raquo;</a>
+                            <a style="float:right;margin-right:10px" href="${url('jobbrowser.views.tasks', job=job.jobId)}">${_('View All Tasks')} &raquo;</a>
                             <h3>
                                 ${_('Recent Tasks')}
                             </h3>
@@ -226,7 +265,8 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                             <td>${_('Status')}</td>
                             <td>${job.status}</td>
                         </tr>
-                            ${rows_for_conf_vars(job.conf_keys)}
+
+                        ${rows_for_conf_vars(job.conf_keys)}
 
                         </tbody>
                     </table>
@@ -237,6 +277,18 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                         <th>${_('Value')}</th>
                         </thead>
                         <tbody>
+                          % if job.is_mr2:
+                            % for line in job.full_job_conf['property']:
+                                <tr>
+                                    <td width="20%">${ line['name'] }</td>
+                                    <td>
+                                        <div class="wordbreak">
+                                        ${ line['value'] }
+                                        </div>
+                                    </td>
+                                </tr>
+                            % endfor
+                          % else:
                             % for key, value in sorted(job.full_job_conf.items()):
                                 <tr>
                                     <td width="20%">${key}</td>
@@ -247,12 +299,17 @@ ${commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                                     </td>
                                 </tr>
                             % endfor
+                          % endif
                         </tbody>
                     </table>
 
                 </div>
                 <div id="counters" class="tab-pane">
-                    ${comps.job_counters(job.counters)}
+                    % if job.is_mr2:
+                      ${ comps.job_counters_mr2(job.counters) }
+                    % else:
+                      ${ comps.job_counters(job.counters) }
+                    % endif
                 </div>
             </div>
         </div>

+ 113 - 0
apps/jobbrowser/src/jobbrowser/templates/job_attempt_logs.mako

@@ -0,0 +1,113 @@
+## Licensed to Cloudera, Inc. under one
+## or more contributor license agreements.  See the NOTICE file
+## distributed with this work for additional information
+## regarding copyright ownership.  Cloudera, Inc. licenses this file
+## to you under the Apache License, Version 2.0 (the
+## "License"); you may not use this file except in compliance
+## with the License.  You may obtain a copy of the License at
+##
+##     http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+<%!
+  from desktop.views import commonheader, commonfooter
+  from django.utils.translation import ugettext as _
+%>
+
+<%namespace name="comps" file="jobbrowser_components.mako" />
+
+${ commonheader(_('Job Browser - Job Attempt: %(attempt_index)s') % {'attempt_index': attempt_index}, "jobbrowser", user) }
+
+<div class="container-fluid">
+    <h1>
+        ${ _('Job Browser') } -
+        ${ _('Job') } <a href="${url('jobbrowser.views.single_job', job=job.jobId)}" title="${_('View this job')}">${ job.jobId_short }</a>
+        ${ _('Attempt: %(attempt_index)s') % {'attempt_index': attempt_index} }
+    </h1>
+    <div class="row-fluid">
+        <div class="span2">
+            <div class="well sidebar-nav">
+                <ul class="nav nav-list">
+                    <li class="nav-header">${ _('Attempt ID') }</li>
+                    <li>${ attempt_index }</li>
+                    <li class="nav-header">${ _('Quick links') }</li>
+                    <li><a href="#stdout">${_('stdout')}</a></li>
+                    <li><a href="#stderr">${_('stderr')}</a></li>
+                    <li><a href="#syslog">${_('syslog')}</a></li>
+                </ul>
+            </div>
+        </div>
+
+        <div class="span10">
+            <ul class="nav nav-tabs">
+                <li class="active">
+                    <a href="#logs" data-toggle="tab">${_('Logs')}</a>
+                </li>
+            </ul>
+
+            <div class="tab-content">
+                <div class="tab-pane active" id="logs">
+                    <h2 id="stdout">${_('stdout')}</h2>
+                    <pre id="stdout-container">
+                        ${_('Loading...')} <img src="/static/art/login-spinner.gif">
+                    </pre>
+
+                    <h2 id="stderr">${_('stderr')}</h2>
+                    <pre id="stderr-container">
+                        ${_('Loading...')} <img src="/static/art/login-spinner.gif">
+                    </pre>
+
+                    <h2 id="syslog">${_('syslog')}</h2>
+                    <pre id="syslog-container">
+                        ${_('Loading...')} <img src="/static/art/login-spinner.gif">
+                    </pre>
+                </div>
+            </div>
+        </div>
+    </div>
+</div>
+
+<script type="text/javascript" charset="utf-8">
+    $(document).ready(function(){
+        $("#metadataTable").dataTable({
+            "bPaginate": false,
+            "bLengthChange": false,
+            "bInfo": false,
+            "bAutoWidth": false,
+            "bFilter": false,
+            "aoColumns": [
+                { "sWidth": "30%" },
+                { "sWidth": "70%" }
+            ]
+        });
+
+        $(".taskCountersTable").dataTable({
+            "bPaginate": false,
+            "bLengthChange": false,
+            "bInfo": false,
+            "bFilter": false,
+            "bAutoWidth": false,
+            "aoColumns": [
+                { "sWidth": "30%" },
+                { "sWidth": "70%" }
+            ]
+        });
+
+        // From 15s to less than 5s display time with async
+        $.get('${ url("jobbrowser.views.job_attempt_logs_json", job=job.jobId, attempt_index=attempt_index, name='syslog') }', function(data) {
+            $('#syslog-container').html(data['log']);
+        });
+        $.get('${ url("jobbrowser.views.job_attempt_logs_json", job=job.jobId, attempt_index=attempt_index, name='stdout') }', function(data) {
+            $('#stdout-container').html(data['log']);
+        });
+		$.get('${ url("jobbrowser.views.job_attempt_logs_json", job=job.jobId, attempt_index=attempt_index, name='stderr') }', function(data) {
+		    $('#stderr-container').html(data['log']);
+		});
+    });
+</script>
+
+${ commonfooter(messages) }

+ 1 - 1
apps/jobbrowser/src/jobbrowser/templates/jobbrowser.html

@@ -68,7 +68,7 @@ limitations under the License.
     <td></td>
     <td>{{clusterstatus.numBlacklistedTrackers}}</td>
     <td></td>
-   </tr>          
+   </tr>
 </table>
 <hr>
 <h2>Scheduling Information</h2>

+ 84 - 5
apps/jobbrowser/src/jobbrowser/templates/jobbrowser_components.mako

@@ -42,6 +42,33 @@
     % endfor
 </%def>
 
+
+<%def name="task_counters_mr2(counters)">
+    <%
+        from jobbrowser.views import format_counter_name
+    %>
+    % for group in counters.get('taskCounterGroup', []):
+        <h3>${ format_counter_name(group['counterGroupName']) }</h3>
+        <table class="taskCountersTable table table-striped table-condensed">
+            <thead>
+            <tr>
+                <th>${_('Counter Name')}</th>
+                <th>${_('Value')}</th>
+            </tr>
+            </thead>
+        <tbody>
+            % for counter in group['counter']:
+            <tr>
+                <td>${format_counter_name(counter['name'])}</td>
+                <td>${counter['value']}</td>
+            </tr>
+            % endfor
+        </tbody>
+        </table>
+    % endfor
+</%def>
+
+
 <%def name="job_counters(counters)">
     <%
         from jobbrowser.views import format_counter_name
@@ -83,6 +110,47 @@
     % endfor
 </%def>
 
+<%def name="job_counters_mr2(counters)">
+    <%
+        from jobbrowser.views import format_counter_name
+    %>
+    % for counter_group in counters.get('counterGroup', []):
+        <h3>${ format_counter_name(counter_group['counterGroupName']) }</h3>
+        <table class="jobCountersTable table table-striped table-condensed">
+            <thead>
+            <tr>
+                <th>${_('Name')}</th>
+                <th>${_('Maps Total')}</th>
+                <th>${_('Reduces Total')}</th>
+                <th>${_('Total')}</th>
+            </tr>
+            </thead>
+        <tbody>
+            % for counter in counter_group['counter']:
+            <%
+                map_count = counter.get('mapCounterValue', 0)
+                reduce_count = counter.get('reduceCounterValue', 0)
+                total_count = counter.get('totalCounterValue', 0)
+            %>
+            <tr>
+                % if not job.is_retired:
+                    <td>${ format_counter_name(counter.get('name', 'n/a')) }</td>
+                    <td>${ map_count }</td>
+                    <td>${ reduce_count }</td>
+                    <td>${ total_count }</td>
+                % else:
+                    <td>N/A</td>
+                    <td>N/A</td>
+                    <td>N/A</td>
+                    <td>N/A</td>
+                % endif
+            </tr>
+            % endfor
+        </tbody>
+        </table>
+    % endfor
+</%def>
+
 <%def name="mr_graph(job)">
     <div>
         ${mr_graph_maps(job)}
@@ -92,14 +160,22 @@
 
 <%def name="mr_graph_maps(job)">
     <div class="progress ${get_bootstrap_class(job, 'progress')}">
-        <div class="bar-label">${job.finishedMaps} / ${job.desiredMaps}</div>
+        % if job.is_mr2:
+            <div class="bar-label">${job.maps_percent_complete}</div>
+        % else:
+            <div class="bar-label">${job.finishedMaps} / ${job.desiredMaps}</div>
+        % endif
         <div class="bar" style="margin-top:-20px;width: ${job.maps_percent_complete}%;"></div>
     </div>
 </%def>
 
 <%def name="mr_graph_reduces(job)">
     <div class="progress ${get_bootstrap_class(job, 'progress')}">
-        <div class="bar-label">${job.finishedReduces} / ${job.desiredReduces}</div>
+        % if job.is_mr2:
+            <div class="bar-label">${job.maps_percent_complete}</div>
+        % else:
+            <div class="bar-label">${job.finishedReduces} / ${job.desiredReduces}</div>
+        % endif
         <div class="bar" style="margin-top:-20px;width: ${job.reduces_percent_complete}%;"></div>
     </div>
 </%def>
@@ -108,7 +184,7 @@
     <%
     additional_class = get_bootstrap_class(job, 'label')
     %>
-    % if job.is_retired:
+    % if job.is_retired and not job.is_mr2:
         <span class="label ${additional_class}"><i class="icon-briefcase icon-white" title="${ _('Retired') }"></i> ${job.status.lower()}</span>
     % else:
         <span class="label ${additional_class}">${job.status.lower()}</span>
@@ -120,9 +196,10 @@
     <%
     additional_class = prefix
     status = job.status.lower()
+
     if status in ('succeeded', 'ok'):
         additional_class += '-success'
-    elif status in ('running', 'prep'):
+    elif status in ('running', 'prep', 'accepted', 'finishing'):
         additional_class += '-warning'
     elif status == 'ready':
         additional_class += '-success'
@@ -133,9 +210,11 @@
             additional_class += '-danger'
         endif
     endif
-    if job.is_retired:
+
+    if job.is_retired and not job.is_mr2:
         additional_class += '-warning'
     endif
+
     return additional_class
     %>
 </%def>

+ 30 - 27
apps/jobbrowser/src/jobbrowser/templates/jobs.mako

@@ -56,11 +56,11 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
     <label class="pull-right">
         &nbsp;
         ${_('Text:')}
-        <input type="text" name="text" title="${_('Text Filter')}" value="${text_filter}" placeholder="${_('Text Filter')}" class="submitter input-large search-query"/>
+        <input type="text" name="text" title="${_('Text Filter')}" value="${ text_filter or '' }" placeholder="${_('Text Filter')}" class="submitter input-large search-query"/>
     </label>
     <label class="pull-right">
         ${_('Username:')}
-        <input type="text" name="user" title="${_('User Name Filter')}" value="${user_filter}" placeholder="${_('User Name Filter')}" class="submitter input-small search-query" />
+        <input type="text" name="user" title="${_('User Name Filter')}" value="${ user_filter or '' }" placeholder="${_('User Name Filter')}" class="submitter input-large search-query" />
     </label>
 </form>
 
@@ -94,30 +94,32 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
         % for job in jobs:
         <tr class="job-row">
             <td data-row-selector-exclude="true">
-                <a href="${ url('jobbrowser.views.job_single_logs', jobid=job.jobId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
+                <a href="${ url('jobbrowser.views.job_single_logs', job=job.jobId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
             </td>
             <td>
-                <a href="${url('jobbrowser.views.single_job', jobid=job.jobId)}" title="${_('View this job')}" data-row-selector="true">${job.jobId_short}</a>
+                <a href="${url('jobbrowser.views.single_job', job=job.jobId)}" title="${_('View this job')}" data-row-selector="true">${job.jobId_short}</a>
             </td>
             <td>
                 ${job.jobName}
             </td>
             <td>
-                <a href="${url('jobbrowser.views.jobs')}?${get_state_link(request, 'state', job.status.lower())}" title="${_('Show only %(status)s jobs') % dict(status=job.status.lower())}" class="nounderline">
+                <a href="${url('jobbrowser.views.jobs')}?${get_state_link(request, 'state', job.status.lower())}"
+                    title="${_('Show only %(status)s jobs') % dict(status=job.status.lower())}" class="nounderline">
                     ${comps.get_status(job)}
                 </a>
             </td>
             <td>
-                <a href="${url('jobbrowser.views.jobs')}?${get_state_link(request, 'user', job.user.lower())}" title="${_('Show only %(status)s jobs') % dict(status=job.user.lower())}">${job.user}</a>
+                <a href="${url('jobbrowser.views.jobs')}?${get_state_link(request, 'user', job.user.lower())}"
+                    title="${_('Show only %(status)s jobs') % dict(status=job.user.lower())}">${job.user}</a>
             </td>
             <td data-sort-value="${job.maps_percent_complete}">
                 % if job.is_retired:
                     <div class="center">${_('N/A')}</div>
                 % else:
-                ${comps.mr_graph_maps(job)}
+                    ${comps.mr_graph_maps(job)}
                 % endif
             </td>
-            <td data-sort-value="${job.reduces_percent_complete}">
+                <td data-sort-value="${job.reduces_percent_complete}">
                 % if job.is_retired:
                     <div class="center">${_('N/A')}</div>
                 % else:
@@ -125,7 +127,7 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
                 % endif
             </td>
             <td>${job.queueName}</td>
-            <td>${job.priority.lower()}</td>
+            <td>${job.priority.lower() or _('N/A')}</td>
             <td data-sort-value="${job.durationInMillis}" data-row-selector-exclude="true">
                 % if job.is_retired:
                     ${_('N/A')}
@@ -135,12 +137,13 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
             </td>
             <td data-sort-value="${job.startTimeMs}">${job.startTimeFormatted}</td>
             <td>
-                % if job.status.lower() == 'running' or job.status.lower() == 'pending':
-                % if request.user.is_superuser or request.user.username == job.user:
-                <a href="#" title="${_('Kill this job')}" kill-action="${url('jobbrowser.views.kill_job', jobid=job.jobId)}?next=${request.get_full_path()|urlencode}" data-row-selector-exclude="true" data-keyboard="true" class="btn btn-mini kill">
-                  <i class="icon-remove"></i> ${_('Kill')}
-                </a>
-                % endif
+                % if (job.status.lower() == 'running' or job.status.lower() == 'pending') and not job.is_mr2:
+                  % if request.user.is_superuser or request.user.username == job.user:
+                    <a href="#" title="${_('Kill this job')}" kill-action="${url('jobbrowser.views.kill_job', job=job.jobId)}?next=${request.get_full_path() | urlencode}"
+                        data-row-selector-exclude="true" data-keyboard="true" class="btn btn-mini kill">
+                      <i class="icon-remove"></i> ${_('Kill')}
+                    </a>
+                  % endif
                 % endif
             </td>
             </tr>
@@ -150,17 +153,17 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
     % endif
 
     % else:
-    ${commonheader(_('Job Browser'), "jobbrowser", user)}
-    <div class="container-fluid">
-    <h1>${_('Welcome to the Job Browser')}</h1>
-    <div>
-        <p>${_("There aren't any jobs running. Let's fix that.")}</p>
-        % if appmanager.get_desktop_module('jobsub') is not None:
-        <a href="/jobsub/">${_('Launch the Job Designer')}</a><br/>
-        % endif
-        % if appmanager.get_desktop_module('beeswax') is not None:
-        <a href="/beeswax/">${_('Launch Beeswax')}</a><br/>
-        % endif
+        ${commonheader(_('Job Browser'), "jobbrowser", user)}
+        <div class="container-fluid">
+        <h1>${_('Welcome to the Job Browser')}</h1>
+        <div>
+            <p>${_("There aren't any jobs running. Let's fix that.")}</p>
+            % if appmanager.get_desktop_module('jobsub') is not None:
+                <a href="/jobsub/">${_('Launch the Job Designer')}</a><br/>
+            % endif
+            % if appmanager.get_desktop_module('beeswax') is not None:
+                <a href="/beeswax/">${_('Launch Beeswax')}</a><br/>
+            % endif
     </div>
     % endif
 </div>
@@ -188,7 +191,7 @@ ${commonheader(_('Job Browser'), "jobbrowser", user)}
             "bLengthChange": false,
             "bFilter": false,
             "bInfo": false,
-            "aaSorting": [[ 0, "desc" ]],
+            "aaSorting": [[1, "desc"]],
             "aoColumns": [
                 {"bSortable": false},
                 null,

+ 9 - 5
apps/jobbrowser/src/jobbrowser/templates/task.mako

@@ -30,7 +30,7 @@ ${commonheader(_('Job Task: %(taskId)s - Job Browser') % dict(taskId=task.taskId
                     <li class="nav-header">${_('Task ID')}</li>
                     <li>${task.taskId_short}</li>
                     <li class="nav-header">${_('Job')}</li>
-                    <li><a href="${url('jobbrowser.views.single_job', jobid=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
+                    <li><a href="${url('jobbrowser.views.single_job', job=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId_short}</a></li>
                     <li class="nav-header">${_('Status')}</li>
                     <li>
                             % if task.state.lower() == 'running' or task.state.lower() == 'pending':
@@ -73,9 +73,9 @@ ${commonheader(_('Job Task: %(taskId)s - Job Browser') % dict(taskId=task.taskId
                         <tbody>
                                 % for attempt in task.attempts:
                                 <tr>
-                                    <td data-row-selector-exclude="true"><a href="${ url('jobbrowser.views.single_task_attempt_logs', jobid=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a></td>
+                                    <td data-row-selector-exclude="true"><a href="${ url('jobbrowser.views.single_task_attempt_logs', job=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a></td>
                                     <td><a title="${_('View this attempt')}"
-                                           href="${ url('jobbrowser.views.single_task_attempt', jobid=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }" data-row-selector="true">${attempt.attemptId_short}</a></td>
+                                           href="${ url('jobbrowser.views.single_task_attempt', job=joblnk.jobId, taskid=task.taskId, attemptid=attempt.attemptId) }" data-row-selector="true">${attempt.attemptId_short}</a></td>
                                     <td>${"%d" % (attempt.progress * 100)}%</td>
                                     <td><span class="status_link ${attempt.state}">${attempt.state}</span></td>
                                     <td><a href="/jobbrowser/trackers/${attempt.taskTrackerId}" class="task_tracker_link">${attempt.taskTrackerId}</a></td>
@@ -108,7 +108,7 @@ ${commonheader(_('Job Task: %(taskId)s - Job Browser') % dict(taskId=task.taskId
                         </tr>
                         <tr>
                             <td>${_('JobId')}</td>
-                            <td><a href="${url('jobbrowser.views.single_job', jobid=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId}</a></td>
+                            <td><a href="${url('jobbrowser.views.single_job', job=joblnk.jobId)}" title="${_('View this job')}">${joblnk.jobId}</a></td>
                         </tr>
                         <tr>
                             <td>${_('State')}</td>
@@ -138,7 +138,11 @@ ${commonheader(_('Job Task: %(taskId)s - Job Browser') % dict(taskId=task.taskId
                     </table>
                 </div>
                 <div id="counters" class="tab-pane">
-                    ${comps.task_counters(task.counters)}
+                    % if task.is_mr2:
+                        ${ comps.task_counters_mr2(task.counters) }
+                    % else:
+                        ${ comps.task_counters(task.counters) }
+                    % endif
                 </div>
             </div>
         </div>

+ 13 - 13
apps/jobbrowser/src/jobbrowser/templates/tasks.mako

@@ -21,17 +21,18 @@
 
 <%namespace name="comps" file="jobbrowser_components.mako" />
 
-${commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=jobid_short), "jobbrowser", user)}
+${ commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short), "jobbrowser", user) }
 
 <%def name="selected(val, state)">
     %   if val is not None and state is not None and val in state:
         selected="true"
     %   endif
 </%def>
+
 <div class="container-fluid">
-    <h1>${_('Task View: Job: %(jobId)s') % dict(jobId=jobid_short)}</h1>
+    <h1>${_('Task View: Job: %(jobId)s') % dict(jobId=job.jobId_short)}</h1>
     <div class="well hueWell">
-        <form method="get" action="/jobbrowser/jobs/${jobid}/tasks">
+        <form method="get" action="${ url('jobbrowser.views.tasks', job=job.jobId) }">
             <b>${_('Filter tasks:')}</b>
 
             <select name="taskstate" class="submitter">
@@ -43,7 +44,6 @@ ${commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=jobid_s
                 <option value="pending" ${selected('pending', taskstate)}>${_('pending')}</option>
             </select>
 
-
             <select name="tasktype" class="submitter">
                 <option value="">${_('All types')}</option>
                 <option value="map" ${selected('map', tasktype)}>${_('maps')}</option>
@@ -57,12 +57,12 @@ ${commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=jobid_s
                 % if tasktext:
                    value="${tasktext}"
                 % endif
-                    />
+            />
         </form>
     </div>
 
 
-    % if len(page.object_list) == 0:
+    % if not page.object_list:
          <p>${_('There were no tasks that match your search criteria.')}</p>
     % else:
         <table class="datatables table table-striped table-condensed">
@@ -84,24 +84,24 @@ ${commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=jobid_s
 	            <tr>
                     <td data-row-selector-exclude="true">
                         %if t.taskAttemptIds:
-                            <a href="${ url('jobbrowser.views.single_task_attempt_logs', jobid=t.jobId, taskid=t.taskId, attemptid=t.taskAttemptIds[-1]) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
+                            <a href="${ url('jobbrowser.views.single_task_attempt_logs', job=t.jobId, taskid=t.taskId, attemptid=t.taskAttemptIds[-1]) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
                         %endif
                     </td>
 	                <td>${t.taskId_short}</td>
 	                <td>${t.taskType}</td>
 	                <td>
-	                   <div class="bar">${"%d" % (t.progress * 100)}%</div>
+	                   <div class="bar">${ "%d" % (t.progress * 100) }%</div>
 	                </td>
 	                <td>
-	                    <a href="${url('jobbrowser.views.tasks', jobid=jobid)}?${get_state_link(request, 'taskstate', t.state.lower())}"
-	                       title="${_('Show only %(state)s tasks') % dict(state=t.state.lower())}"
-	                       class="${t.state.lower()}">${t.state.lower()}
+	                    <a href="${ url('jobbrowser.views.tasks', job=job.jobId) }?${ get_state_link(request, 'taskstate', t.state.lower()) }"
+	                       title="${ _('Show only %(state)s tasks') % dict(state=t.state.lower()) }"
+	                       class="${ t.state.lower() }">${ t.state.lower() }
 	                    </a>
 	                </td>
 	                <td>${t.mostRecentState}</td>
 	                <td>${t.execStartTimeFormatted}</td>
 	                <td>${t.execFinishTimeFormatted}</td>
-	                <td><a href="/jobbrowser/jobs/${jobid}/tasks/${t.taskId}" data-row-selector="true">${_('Attempts')}</a></td>
+	                <td><a href="${ url('jobbrowser.views.single_task', job=job.jobId, taskid=t.taskId) }" data-row-selector="true">${_('Attempts')}</a></td>
 	            </tr>
 	            %endfor
 	        </tbody>
@@ -127,4 +127,4 @@ ${commonheader(_('Task View: Job: %(jobId)s - Job Browser') % dict(jobId=jobid_s
     });
 </script>
 
-${commonfooter(messages)}
+${ commonfooter(messages) }

+ 206 - 1
apps/jobbrowser/src/jobbrowser/tests.py

@@ -27,6 +27,9 @@ from nose.tools import assert_true, assert_false, assert_equal
 
 from desktop.lib.django_test_util import make_logged_in_client
 from desktop.lib.test_utils import grant_access
+from hadoop import cluster
+from hadoop.conf import YARN_CLUSTERS
+from hadoop.yarn import resource_manager_api, mapreduce_api, history_server_api
 from jobsub.models import OozieDesign, CheckForSetup
 from liboozie.oozie_api_test import OozieServerProvider
 
@@ -310,7 +313,7 @@ class TestJobBrowserWithHadoop(unittest.TestCase, OozieServerProvider):
     # All jobs page and fetch job ID
     # Taking advantage of the fact new jobs are at the top of the list!
     response = self.client.get('/jobbrowser/jobs/')
-    assert_true(hadoop_job_id_short in response.content)
+    assert_true(hadoop_job_id_short in response.content, response.content)
 
     # Make sure job succeeded
     response = self.client.get('/jobbrowser/jobs/?state=completed')
@@ -391,3 +394,205 @@ class TestJobBrowserWithHadoop(unittest.TestCase, OozieServerProvider):
     # Test job single logs page
     response = self.client.get('/jobbrowser/jobs/%s/single_logs' % (hadoop_job_id))
     assert_true('syslog' in response.content)
+
+
+
+class TestMapReduce2:
+
+  def setUp(self):
+    # Beware: Monkey patching
+    if not hasattr(resource_manager_api, 'old_get_resource_manager_api'):
+      resource_manager_api.old_get_resource_manager = resource_manager_api.get_resource_manager
+    if not hasattr(resource_manager_api, 'old_get_mapreduce_api'):
+      mapreduce_api.old_get_mapreduce_api = mapreduce_api.get_mapreduce_api
+    if not hasattr(history_server_api, 'old_get_history_server_api'):
+      history_server_api.old_get_history_server_api = history_server_api.get_history_server_api
+
+    resource_manager_api.get_resource_manager = lambda: MockResourceManagerApi()
+    mapreduce_api.get_mapreduce_api = lambda: MockMapreduceApi()
+    history_server_api.get_history_server_api = lambda: HistoryServerApi()
+
+
+    self.c = make_logged_in_client(is_superuser=False)
+    grant_access("test", "test", "jobbrowser")
+
+    self.finish = YARN_CLUSTERS['default'].SUBMIT_TO.set_for_testing(True)
+    assert_true(cluster.is_yarn())
+
+  def tearDown(self):
+    resource_manager_api.get_resource_manager = getattr(resource_manager_api, 'old_get_resource_manager')
+    mapreduce_api.get_mapreduce_api = getattr(mapreduce_api, 'old_get_mapreduce_api')
+    history_server_api.get_history_server_api = getattr(history_server_api, 'old_get_history_server_api')
+
+    self.finish()
+
+
+  def test_jobs(self):
+    response = self.c.get('/jobbrowser/')
+    assert_equal(len(response.context['jobs']), 2)
+
+    # state=running comes from the API and so can't be mocked
+
+    response = self.c.get('/jobbrowser/jobs/?text=W=MapReduce-copy2')
+    assert_equal(len(response.context['jobs']), 1)
+
+  def test_running_job(self):
+    response = self.c.get('/jobbrowser/jobs/application_1356251510842_0054')
+    assert_equal(response.context['job'].jobId, 'application_1356251510842_0054')
+
+    response = self.c.get('/jobbrowser/jobs/job_1356251510842_0054')
+    assert_false('job' in response.context)
+
+  def test_finished_job(self):
+    response = self.c.get('/jobbrowser/jobs/application_1356251510842_0009')
+    assert_equal(response.context['job'].jobId, 'job_1356251510842_0009')
+
+    response = self.c.get('/jobbrowser/jobs/job_1356251510842_0009')
+    assert_equal(response.context['job'].jobId, 'job_1356251510842_0009')
+
+
+class MockResourceManagerApi:
+
+  def __init__(self, oozie_url=None): pass
+
+  def apps(self, **kwargs):
+    return {
+      u'apps':
+        {u'app': [
+           # RUNNING application_1356251510842_0054
+           {u'finishedTime': 1356961070119, u'name': u'oozie:launcher:T=map-reduce:W=MapReduce-copy:A=Sleep:ID=0000004-121223003201296-oozie-oozi-W',
+            u'amContainerLogs': u'http://runreal:8042/node/containerlogs/container_1356251510842_0054_01_000001/romain', u'clusterId': 1356251510842,
+            u'trackingUrl': u'http://localhost:8088/proxy/application_1356251510842_0054/jobhistory/job/job_1356251510842_0054', u'amHostHttpAddress': u'runreal:8042',
+            u'startedTime': 1356961057225, u'queue': u'default', u'state': u'RUNNING', u'elapsedTime': 12894, u'finalStatus': u'UNDEFINED', u'diagnostics': u'',
+            u'progress': 100.0, u'trackingUI': u'History', u'id': u'application_1356251510842_0054', u'user': u'romain'},
+           # FINISHED application_1356251510842_0009
+           {u'finishedTime': 1356467118570, u'name': u'oozie:action:T=map-reduce:W=MapReduce-copy2:A=Sleep:ID=0000002-121223003201296-oozie-oozi-W',
+            u'amContainerLogs': u'http://runreal:8042/node/containerlogs/container_1356251510842_0009_01_000001/romain', u'clusterId': 1356251510842,
+            u'trackingUrl': u'http://localhost:8088/proxy/application_1356251510842_0009/jobhistory/job/job_1356251510842_0009', u'amHostHttpAddress': u'runreal:8042',
+            u'startedTime': 1356467081121, u'queue': u'default', u'state': u'FINISHED', u'elapsedTime': 37449, u'finalStatus': u'SUCCEEDED', u'diagnostics': u'',
+            u'progress': 100.0, u'trackingUI': u'History', u'id': u'application_1356251510842_0009', u'user': u'romain'}]
+      }
+    }
+
+
+class MockMapreduce2Api(object):
+  """
+  MockMapreduceApi and HistoryServerApi are very similar and inherit from it.
+  """
+
+  def __init__(self, oozie_url=None): pass
+
+  def tasks(self, job_id):
+    return {u'tasks': {u'task': [{u'finishTime': 1357153330271, u'successfulAttempt': u'attempt_1356251510842_0062_m_000000_0', u'elapsedTime': 1901, u'state': u'SUCCEEDED',
+                                  u'startTime': 1357153328370, u'progress': 100.0, u'type': u'MAP', u'id': u'task_1356251510842_0062_m_000000'},
+                                 {u'finishTime': 0, u'successfulAttempt': u'', u'elapsedTime': 0, u'state': u'SCHEDULED', u'startTime': 1357153326322, u'progress': 0.0,
+                                  u'type': u'REDUCE', u'id': u'task_1356251510842_0062_r_000000'}]}}
+  def conf(self, job_id):
+    return {
+      "conf" : {
+        "path" : "hdfs://host.domain.com:9000/user/user1/.staging/job_1326232085508_0004/job.xml",
+        "property" : [
+           {
+              "value" : "/home/hadoop/hdfs/data",
+              "name" : "dfs.datanode.data.dir"
+           },]
+         }
+    }
+
+  def job_attempts(self, job_id):
+    return {
+       "jobAttempts" : {
+          "jobAttempt" : [
+             {
+                "nodeId" : "host.domain.com:8041",
+                "nodeHttpAddress" : "host.domain.com:8042",
+                "startTime" : 1326238773493,
+                "id" : 1,
+                "logsLink" : "http://host.domain.com:8042/node/containerlogs/container_1326232085508_0004_01_000001",
+                "containerId" : "container_1326232085508_0004_01_000001"
+             }
+          ]
+       }
+    }
+
+  def task_attempts(self, job_id, task_id):
+    return {
+       "taskAttempts" : {
+          "taskAttempt" : [
+             {
+                "elapsedMergeTime" : 47,
+                "shuffleFinishTime" : 1326238780052,
+                "assignedContainerId" : "container_1326232085508_0004_01_000003",
+                "progress" : 100,
+                "elapsedTime" : 0,
+                "state" : "RUNNING",
+                "elapsedShuffleTime" : 2592,
+                "mergeFinishTime" : 1326238780099,
+                "rack" : "/98.139.92.0",
+                "elapsedReduceTime" : 0,
+                "nodeHttpAddress" : "host.domain.com:8042",
+                "type" : "REDUCE",
+                "startTime" : 1326238777460,
+                "id" : "attempt_1326232085508_4_4_r_0_0",
+                "finishTime" : 0
+             }
+          ]
+       }
+    }
+
+  def counters(self, job_id):
+    return {
+       "jobCounters" : {
+          "id" : "job_1326232085508_4_4",
+          "counterGroup" : [
+             {
+                "counterGroupName" : "org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter",
+                "counter" : [
+                   {
+                      "reduceCounterValue" : 0,
+                      "mapCounterValue" : 0,
+                      "totalCounterValue" : 0,
+                      "name" : "BYTES_READ"
+                   }
+                ]
+             },
+             {
+                "counterGroupName" : "org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter",
+                "counter" : [
+                   {
+                      "reduceCounterValue" : 0,
+                      "mapCounterValue" : 0,
+                      "totalCounterValue" : 0,
+                      "name" : "BYTES_WRITTEN"
+                   }
+                ]
+             }
+          ]
+       }
+    }
+
+
+class MockMapreduceApi(MockMapreduce2Api):
+  def job(self, user, job_id):
+    if '1356251510842_0009' not in job_id:
+      job = {u'job': {u'reducesCompleted': 0, u'mapsRunning': 1, u'id': u'job_1356251510842_0054', u'successfulReduceAttempts': 0, u'successfulMapAttempts': 0,
+                      u'uberized': False, u'reducesTotal': 1, u'elapsedTime': 3426, u'mapsPending': 0, u'state': u'RUNNING', u'failedReduceAttempts': 0,
+                      u'mapsCompleted': 0, u'killedMapAttempts': 0, u'killedReduceAttempts': 0, u'runningReduceAttempts': 0, u'failedMapAttempts': 0, u'mapsTotal': 1,
+                      u'user': u'romain', u'startTime': 1357152972886, u'reducesPending': 1, u'reduceProgress': 0.0, u'finishTime': 0,
+                      u'name': u'select avg(salary) from sample_07(Stage-1)', u'reducesRunning': 0, u'newMapAttempts': 0, u'diagnostics': u'', u'mapProgress': 0.0,
+                      u'runningMapAttempts': 1, u'newReduceAttempts': 1}}
+      job['job']['id'] = job_id
+      return job
+
+
+class HistoryServerApi(MockMapreduce2Api):
+
+  def __init__(self, oozie_url=None): pass
+
+  def job(self, user, job_id):
+    if '1356251510842_0054' not in job_id:
+      return {u'job': {u'reducesCompleted': 1, u'avgMapTime': 1798, u'avgMergeTime': 1479, u'id': u'job_1356251510842_0009', u'successfulReduceAttempts': 1,
+                     u'successfulMapAttempts': 2, u'uberized': False, u'reducesTotal': 1, u'state': u'SUCCEEDED', u'failedReduceAttempts': 0, u'mapsCompleted': 2,
+                     u'killedMapAttempts': 0, u'diagnostics': u'', u'mapsTotal': 2, u'user': u'romain', u'startTime': 1357151916268, u'avgReduceTime': 137,
+                     u'finishTime': 1357151923925, u'name': u'oozie:action:T=map-reduce:W=MapReduce-copy:A=Sleep:ID=0000004-121223003201296-oozie-oozi-W',
+                     u'avgShuffleTime': 1421, u'queue': u'default', u'killedReduceAttempts': 0, u'failedMapAttempts': 0}}

+ 16 - 13
apps/jobbrowser/src/jobbrowser/urls.py

@@ -20,22 +20,25 @@ from django.conf.urls.defaults import patterns, url
 urlpatterns = patterns('jobbrowser.views',
   # "Default"
   url(r'^$', 'jobs'),
-  url(r'^trackers$', 'trackers', name='trackers'),
-  url(r'^trackers/(?P<trackerid>.+)$', 'single_tracker', name='single_tracker'),
   url(r'^jobs/$', 'jobs', name='jobs'),
-  url(r'^dock_jobs/$', 'dock_jobs', name='dock_jobs'),
-  url(r'^jobs/(?P<jobid>\w+)$','single_job', name='single_job'),
-  url(r'^jobs/(?P<jobid>\w+)/counters$', 'job_counters', name='job_counters'),
-  url(r'^jobs/(?P<jobid>\w+)/kill$', 'kill_job', name='kill_job'),
-  url(r'^jobs/(?P<jobid>\w+)/setpriority$', 'set_job_priority', name='set_job_priority'),
-  url(r'^jobs/(?P<jobid>\w+)/single_logs$', 'job_single_logs', name='job_single_logs'),
-  url(r'^jobs/(?P<jobid>\w+)/tasks$','tasks', name='tasks'),
-  url(r'^jobs/(?P<jobid>\w+)/tasks/(?P<taskid>\w+)$', 'single_task', name='single_task'),
-  url(r'^jobs/(?P<jobid>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)$', 'single_task_attempt', name='single_task_attempt'),
-  url(r'^jobs/(?P<jobid>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)/counters$', 'task_attempt_counters', name='task_attempt_counters'),
-  url(r'^jobs/(?P<jobid>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)/logs$', 'single_task_attempt_logs', name='single_task_attempt_logs'),
+  url(r'^jobs/(?P<job>\w+)$','single_job', name='single_job'),
+  url(r'^jobs/(?P<job>\w+)/counters$', 'job_counters', name='job_counters'),
+  url(r'^jobs/(?P<job>\w+)/kill$', 'kill_job', name='kill_job'),
+  url(r'^jobs/(?P<job>\w+)/setpriority$', 'set_job_priority', name='set_job_priority'), #? used
+  url(r'^jobs/(?P<job>\w+)/single_logs$', 'job_single_logs', name='job_single_logs'),
+  url(r'^jobs/(?P<job>\w+)/job_attempt_logs/(?P<attempt_index>\d+)$', 'job_attempt_logs', name='job_attempt_logs'),
+  url(r'^jobs/(?P<job>\w+)/job_attempt_logs_json/(?P<attempt_index>\d+)/(?P<name>\w+)?/(?P<offset>\d+)?$', 'job_attempt_logs_json', name='job_attempt_logs_json'),
+  url(r'^jobs/(?P<job>\w+)/tasks$','tasks', name='tasks'),
+  url(r'^jobs/(?P<job>\w+)/tasks/(?P<taskid>\w+)$', 'single_task', name='single_task'), # TODO s/single// ?
+  url(r'^jobs/(?P<job>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)$', 'single_task_attempt', name='single_task_attempt'),
+  url(r'^jobs/(?P<job>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)/counters$', 'task_attempt_counters', name='task_attempt_counters'),
+  url(r'^jobs/(?P<job>\w+)/tasks/(?P<taskid>\w+)/attempts/(?P<attemptid>\w+)/logs$', 'single_task_attempt_logs', name='single_task_attempt_logs'),
   url(r'^jobs/(\w+)/tasks/(\w+)/attempts/(?P<attemptid>\w+)/kill$', 'kill_task_attempt', name='kill_task_attempt'),
+  # Unused
   url(r'^clusterstatus$', 'clusterstatus', name='clusterstatus'),
   url(r'^queues$', 'queues', name='queues'),
   url(r'^jobbrowser$', 'jobbrowser', name='jobbrowser'),
+  url(r'^trackers$', 'trackers', name='trackers'),
+  url(r'^trackers/(?P<trackerid>.+)$', 'single_tracker', name='single_tracker'),
+  url(r'^dock_jobs/$', 'dock_jobs', name='dock_jobs'),
 )

+ 154 - 257
apps/jobbrowser/src/jobbrowser/views.py

@@ -14,38 +14,34 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#
-# Implements simple jobbrowser api
-#
+
 import re
 import time
 import logging
 import string
 from urllib import quote_plus
+from lxml import html
 
-from desktop.lib.paginator import Paginator
-from desktop.lib.django_util import render_json, render, copy_query_dict
-from desktop.lib.exceptions import MessageException
-from desktop.lib.exceptions_renderable import PopupException
-from desktop.lib.conf import coerce_bool
+try:
+  import json
+except ImportError:
+  import simplejson as json
 
-from django.http import HttpResponseRedirect
+from django.http import HttpResponseRedirect, HttpResponse
 from django.utils.functional import wraps
+from django.utils.translation import ugettext as _
 
 from desktop.log.access import access_warn, access_log_level
+from desktop.lib.rest.http_client import RestException
+from desktop.lib.django_util import render_json, render, copy_query_dict
+from desktop.lib.exceptions import MessageException
+from desktop.lib.exceptions_renderable import PopupException
 from desktop.views import register_status_bar_view
 from hadoop.api.jobtracker.ttypes import ThriftJobPriority, TaskTrackerNotFoundException, ThriftJobState
 
-
 from jobbrowser import conf
-from jobbrowser.models import Job, JobLinkage, TaskList, Tracker, Cluster
-
-from django.utils.translation import ugettext as _
-
-##################################
-## View end-points
-
-__DEFAULT_OBJ_PER_PAGINATION = 10
+from jobbrowser.api import get_api
+from jobbrowser.models import Job, JobLinkage, Tracker, Cluster
 
 
 def check_job_permission(view_func):
@@ -54,22 +50,40 @@ def check_job_permission(view_func):
   Assumes that the wrapped function takes a 'jobid' param.
   """
   def decorate(request, *args, **kwargs):
-    jobid = kwargs['jobid']
-    job = Job.from_id(jt=request.jt, jobid=jobid)
+    jobid = kwargs['job']
+    try:
+      job = get_api(request.user, request.jt).get_job(jobid=jobid)
+    except Exception, e:
+      raise PopupException(_('Could not find job %s. The job might not be running yet.') % jobid, detail=e)
     if not conf.SHARE_JOBS.get() and not request.user.is_superuser \
       and job.user != request.user.username:
-      raise PopupException(_("You don't have the permissions to access"
-                             " job %(id)s.") % dict(id=jobid))
+      raise PopupException(_("You don't have the permissions to access job %(id)s.") % {'id': jobid})
+    kwargs['job'] = job
     return view_func(request, *args, **kwargs)
   return wraps(view_func)(decorate)
 
-@check_job_permission
-def single_job(request, jobid):
-  """
-  We get here from /jobs/jobid
-  """
-  job = Job.from_id(jt=request.jt, jobid=jobid)
 
+def jobs(request):
+  user = request.GET.get('user', request.user.username)
+  state = request.GET.get('state')
+  text = request.GET.get('text')
+  retired = request.GET.get('retired')
+
+  jobs = get_api(request.user, request.jt).get_jobs(user=request.user, username=user, state=state, text=text, retired=retired)
+
+  return render('jobs.mako', request, {
+    'jobs': jobs,
+    'request': request,
+    'state_filter': state,
+    'user_filter': user,
+    'text_filter': text,
+    'retired': retired,
+    'filtered': not (state == 'all' and user == '' and text == '')
+  })
+
+
+@check_job_permission
+def single_job(request, job):
   def cmp_exec_time(task1, task2):
     return cmp(task1.execStartTimeMs, task2.execStartTimeMs)
 
@@ -78,66 +92,25 @@ def single_job(request, jobid):
   recent_tasks = job.filter_tasks(task_states=('running', 'succeeded',))
   recent_tasks.sort(cmp_exec_time, reverse=True)
 
-  return render("job.mako", request, {
+  return render('job.mako', request, {
     'request': request,
     'job': job,
     'failed_tasks': failed_tasks[:5],
-    'recent_tasks': recent_tasks[:5]
+    'recent_tasks': recent_tasks[:5],
   })
 
-@check_job_permission
-def job_counters(request, jobid):
-  """
-  We get here from /jobs/jobid/counters
-  """
-  job = Job.from_id(jt=request.jt, jobid=jobid)
-  return render("counters.html", request, {"counters":job.counters})
-
-def jobs(request):
-  """
-  We get here from /jobs?filterargs
-  """
-  check_permission = not conf.SHARE_JOBS.get() and not request.user.is_superuser
-
-  user = request.GET.get('user', request.user.username)
-  filters = {}
-  if user != '':
-    filters['user'] = user
 
-  jobs = get_matching_jobs(request, check_permission, **filters)
-
-  matching_jobs = sort_if_necessary(request, jobs)
-  state = request.GET.get('state', 'all')
-  text = request.GET.get('text', '')
-  retired = request.GET.get('retired', '')
-
-  return render("jobs.mako", request, {
-    'jobs': matching_jobs,
-    'request': request,
-    'state_filter': state,
-    'user_filter': user,
-    'text_filter': text,
-    'retired': retired,
-    'filtered': not (state == 'all' and user == '' and text == '')
-  })
-
-def dock_jobs(request):
-  username = request.user.username
-  matching_jobs = get_job_count_by_state(request, username)
-  return render("jobs_dock_info.mako", request, {
-    'jobs':matching_jobs
-  }, force_template=True)
-register_status_bar_view(dock_jobs)
+@check_job_permission
+def job_counters(request, job):
+  return render("counters.html", request, {"counters": job.counters})
 
 
 @access_log_level(logging.WARN)
-def kill_job(request, jobid):
-  """
-  We get here from /jobs/jobid/kill
-  """
+@check_job_permission
+def kill_job(request, job):
   if request.method != "POST":
     raise Exception(_("kill_job may only be invoked with a POST (got a %(method)s).") % dict(method=request.method))
-  job = Job.from_id(jt=request.jt, jobid=jobid)
+
   if job.user != request.user.username and not request.user.is_superuser:
     access_warn(request, _('Insufficient permission'))
     raise MessageException(_("Permission denied.  User %(username)s cannot delete user %(user)s's job.") %
@@ -146,7 +119,7 @@ def kill_job(request, jobid):
   job.kill()
   cur_time = time.time()
   while time.time() - cur_time < 15:
-    job = Job.from_id(jt=request.jt, jobid=jobid)
+    job = Job.from_id(jt=request.jt, jobid=job.jobId)
 
     if job.status not in ["RUNNING", "QUEUED"]:
       if request.REQUEST.get("next"):
@@ -154,16 +127,50 @@ def kill_job(request, jobid):
       else:
         raise MessageException("Job Killed")
     time.sleep(1)
-    job = Job.from_id(jt=request.jt, jobid=jobid)
+    job = Job.from_id(jt=request.jt, jobid=job.jobId)
 
   raise Exception(_("Job did not appear as killed within 15 seconds"))
 
+
 @check_job_permission
-def job_single_logs(request, jobid):
-  """
-  We get here from /jobs/jobid/logs
-  """
-  job = Job.from_id(jt=request.jt, jobid=jobid)
+def job_attempt_logs(request, job, attempt_index=0):
+  return render("job_attempt_logs.mako", request, {
+    "attempt_index": attempt_index,
+    "job": job,
+  })
+
+
+@check_job_permission
+def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=0):
+  """For async log retrieval as Yarn servers are very slow"""
+
+  try:
+    attempt_index = int(attempt_index)
+    attempt = job.job_attempts['jobAttempt'][attempt_index]
+    log_link = attempt['logsLink']
+  except (KeyError, RestException), e:
+    raise KeyError(_("Cannot find job attempt '%(id)s'") % {'id': job.jobId}, e)
+
+  link = '/%s/' % name
+  if offset >= 0:
+    link += '?start=%d' % offset
+
+  try:
+    log = html.parse(log_link + link).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
+  except Exception, e:
+    log = _('Failed to retrieve log: %s') % e
+
+  response = {'log': log}
+
+  return HttpResponse(json.dumps(response), mimetype="application/json")
+
+
+
+@check_job_permission
+def job_single_logs(request, job):
+
+  if job.is_mr2:
+    return job_attempt_logs(request, job=job.jobId)
 
   def cmp_exec_time(task1, task2):
     return cmp(task1.execStartTimeMs, task2.execStartTimeMs)
@@ -181,14 +188,14 @@ def job_single_logs(request, jobid):
       task = recent_tasks[0]
 
   if task is None:
-    raise PopupException(_("No tasks found for job %(id)s") % dict(id=jobid))
+    raise PopupException(_("No tasks found for job %(id)s") % {'id': job.jobId})
 
-  return single_task_attempt_logs(request, **{'jobid': jobid, 'taskid': task.taskId, 'attemptid': task.taskAttemptIds[-1]})
+  return single_task_attempt_logs(request, **{'job': job.jobId, 'taskid': task.taskId, 'attemptid': task.taskAttemptIds[-1]})
 
 @check_job_permission
-def tasks(request, jobid):
+def tasks(request, job):
   """
-  We get here from /jobs/jobid/tasks?filterargs, with the options being:
+  We get here from /jobs/job/tasks?filterargs, with the options being:
     page=<n>            - Controls pagination. Defaults to 1.
     tasktype=<type>     - Type can be one of hadoop.job_tracker.VALID_TASK_TYPES
                           ("map", "reduce", "job_cleanup", "job_setup")
@@ -196,42 +203,30 @@ def tasks(request, jobid):
                           ("succeeded", "failed", "running", "pending", "killed")
     tasktext=<text>     - Where <text> is a string matching info on the task
   """
-  # Get the filter parameters
   ttypes = request.GET.get('tasktype')
   tstates = request.GET.get('taskstate')
   ttext = request.GET.get('tasktext')
-  task_types = None
-  if ttypes:
-    task_types = set(ttypes.split(','))
-  task_states = None
-  if tstates:
-    task_states = set(tstates.split(','))
-
   pagenum = int(request.GET.get('page', 1))
-  if pagenum < 0:
-    pagenum = 1
+  pagenum = pagenum > 0 and pagenum or 1
+
+  filters = {
+    'task_types': ttypes and set(ttypes.split(',')) or None,
+    'task_states': tstates and set(tstates.split(',')) or None,
+    'task_text': ttext,
+    'pagenum': pagenum,
+  }
 
-  # Fetch the list of tasks
-  task_list = TaskList.select(request.jt,
-                              jobid,
-                              task_types,
-                              task_states,
-                              ttext,
-                              __DEFAULT_OBJ_PER_PAGINATION,
-                              __DEFAULT_OBJ_PER_PAGINATION * (pagenum - 1))
+  jt = get_api(request.user, request.jt)
 
-  paginator = Paginator(task_list, __DEFAULT_OBJ_PER_PAGINATION, total=task_list.numTotalTasks)
-  page = paginator.page(pagenum)
+  task_list = jt.get_tasks(job.jobId, **filters)
+  page = jt.paginate_task(task_list, pagenum)
 
-  # We need to pass the parameters back to the template to generate links
-  filter_params = copy_query_dict(
-        request.GET, ('tasktype', 'taskstate', 'tasktext')).urlencode()
+  filter_params = copy_query_dict(request.GET, ('tasktype', 'taskstate', 'tasktext')).urlencode()
 
   return render("tasks.mako", request, {
     'request': request,
     'filter_params': filter_params,
-    'jobid':jobid,
-    'jobid_short': get_shorter_id(jobid),
+    'job': job,
     'page': page,
     'tasktype': ttypes,
     'taskstate': tstates,
@@ -240,11 +235,10 @@ def tasks(request, jobid):
 
 
 @check_job_permission
-def single_task(request, jobid, taskid):
-  """
-  We get here from /jobs/jobid/tasks/taskid
-  """
-  job_link = JobLinkage(request.jt, jobid)
+def single_task(request, job, taskid):
+  jt = get_api(request.user, request.jt)
+
+  job_link = jt.get_job_link(job.jobId)
   task = job_link.get_task(taskid)
 
   return render("task.mako", request, {
@@ -253,16 +247,16 @@ def single_task(request, jobid, taskid):
   })
 
 @check_job_permission
-def single_task_attempt(request, jobid, taskid, attemptid):
-  """
-  We get here from /jobs/jobid/tasks/taskid/attempts/attemptid
-  """
-  job_link = JobLinkage(request.jt, jobid)
+def single_task_attempt(request, job, taskid, attemptid):
+  jt = get_api(request.user, request.jt)
+
+  job_link = jt.get_job_link(job.jobId)
   task = job_link.get_task(taskid)
+
   try:
     attempt = task.get_attempt(attemptid)
-  except KeyError:
-    raise KeyError(_("Cannot find attempt '%(id)s' in task") % dict(id=attemptid))
+  except (KeyError, RestException), e:
+    raise PopupException(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
 
   return render("attempt.mako", request,
     {
@@ -273,16 +267,20 @@ def single_task_attempt(request, jobid, taskid, attemptid):
     })
 
 @check_job_permission
-def single_task_attempt_logs(request, jobid, taskid, attemptid):
-  """
-  We get here from /jobs/jobid/tasks/taskid/attempts/attemptid/logs
-  """
-  job_link = JobLinkage(request.jt, jobid)
+def single_task_attempt_logs(request, job, taskid, attemptid):
+  jt = get_api(request.user, request.jt)
+
+  job_link = jt.get_job_link(job.jobId)
+
+  if job_link.is_mr2:
+    return job_attempt_logs(request, job=job.jobId)
+
   task = job_link.get_task(taskid)
+
   try:
     attempt = task.get_attempt(attemptid)
-  except KeyError:
-    raise KeyError(_("Cannot find attempt '%(id)s' in task") % dict(id=attemptid))
+  except (KeyError, RestException), e:
+    raise KeyError(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
 
   try:
     # Add a diagnostic log
@@ -294,22 +292,22 @@ def single_task_attempt_logs(request, jobid, taskid, attemptid):
     # Four entries,
     # for diagnostic, stdout, stderr and syslog
     logs = [ _("Failed to retrieve log. TaskTracker not found.") ] * 4
-  return render("attempt_logs.mako", request,
-    {
-      "attempt":attempt,
-      "taskid":taskid,
+
+  return render("attempt_logs.mako", request, {
+      "attempt": attempt,
+      "taskid": taskid,
       "joblnk": job_link,
       "task": task,
       "logs": logs
     })
 
 @check_job_permission
-def task_attempt_counters(request, jobid, taskid, attemptid):
+def task_attempt_counters(request, job, taskid, attemptid):
   """
   We get here from /jobs/jobid/tasks/taskid/attempts/attemptid/counters
   (phew!)
   """
-  job_link = JobLinkage(request.jt, jobid)
+  job_link = JobLinkage(request.jt, job.jobId)
   task = job_link.get_task(taskid)
   attempt = task.get_attempt(attemptid)
   counters = {}
@@ -321,6 +319,7 @@ def task_attempt_counters(request, jobid, taskid, attemptid):
 def kill_task_attempt(request, attemptid):
   """
   We get here from /jobs/jobid/tasks/taskid/attempts/attemptid/kill
+  TODO: security
   """
   ret = request.jt.kill_task_attempt(request.jt.thriftattemptid_from_string(attemptid))
   return render_json({})
@@ -329,7 +328,7 @@ def trackers(request):
   """
   We get here from /trackers
   """
-  trackers = sort_if_necessary(request, get_tasktrackers(request))
+  trackers = get_tasktrackers(request)
 
   return render("tasktrackers.mako", request, {'trackers':trackers})
 
@@ -353,12 +352,12 @@ def queues(request):
   return render("queues.html", request, { "queuelist" : request.jt.queues()})
 
 @check_job_permission
-def set_job_priority(request, jobid):
+def set_job_priority(request, job):
   """
-  We get here from /jobs/jobid/setpriority?priority=PRIORITY
+  We get here from /jobs/job/setpriority?priority=PRIORITY
   """
   priority = request.GET.get("priority")
-  jid = request.jt.thriftjobid_from_string(jobid)
+  jid = request.jt.thriftjobid_from_string(job.jobId)
   request.jt.set_job_priority(jid, ThriftJobPriority._NAMES_TO_VALUES[priority])
   return render_json({})
 
@@ -369,7 +368,7 @@ def make_substitutions(conf):
   Substitute occurences of ${foo} with conf[foo], recursively, in all the values
   of the conf dict.
 
-  Note that the Java code may also substitute Java properties in, which 
+  Note that the Java code may also substitute Java properties in, which
   this code does not have.
   """
   r = re.compile(CONF_VARIABLE_REGEX)
@@ -411,13 +410,7 @@ def format_counter_name(s):
   return string.capwords(re.sub('_', ' ', splitCamels(s)).lower())
 
 
-def sort_if_necessary(request, items):
-  if request.GET.get("sortkey"):
-    items.sort(key=lambda x: getattr(x, request.GET.get("sortkey")), reverse=request.GET.has_key("sortrev"))
-  return items
-
-def get_state_link(request, option=None, val='',
-                    VALID_OPTIONS = ("state", "user", "text", "taskstate")):
+def get_state_link(request, option=None, val='', VALID_OPTIONS = ("state", "user", "text", "taskstate")):
   """
     constructs the query string for the state of the current query for the jobs page.
     pass in the request, and an optional option/value pair; these are used for creating
@@ -438,67 +431,17 @@ def get_state_link(request, option=None, val='',
   return "&".join([ "%s=%s" % (key, quote_plus(value)) for key, value in states.iteritems() ])
 
 
-def _filter_jobs_by_req(joblist, request, **kwargs):
-  """
-  Unpacks filter arguments from the request object and optional
-  keyword arguments, and supplies the resulting filter to _filter_jobs.
-  """
-  args = {}
-  for x in ["jobid_exact", "jobid_substr", "pools", "user", "tasks", "text"]:
-    if x in kwargs:
-      args[x] = kwargs[x]
-    else:
-      args[x] = request.GET.get(x)
-  return _filter_jobs(joblist, **args)
-
-
-def _filter_jobs(jobs, jobid_exact=None, jobid_substr=None, pools=None, user=None, tasks=None, text=None):
-  # TODO(henry): this naive version can be replaced with something
-  # more flexible. (i.e. use getattr with a dict of values, check
-  # the type and do the right kind of test)
-  """
-  Return the set in jobs that match the supplied parameters (any of which may be
-  None). If jobid is supplied will only return exact id matches if exactid = True.
-
-  All other parameters are substring matched.
-  """
-  def predicate(job):
-    """
-    Return True if a ThriftJobInProgress structure matches the supplied filters.
-
-    If a filter argument is None, everything matches it.
-    """
-    if jobid_exact and jobid_exact != job.jobID.asString:
-      return False
-    if jobid_substr and jobid_substr not in job.jobID.asString:
-      return False
-    if pools and pools not in job.profile.queueName:
-      return False
-    if user and user not in job.profile.user:
-      return False
-    if tasks and not True: # TODO: figure out what Nutron wants to happen here
-      return False
-    if text:
-      search = text.lower()
-      # These fields are chosen to match those displayed by the JT UI
-      saw_text = False
-      for t in [job.profile.user,
-                job.profile.name,
-                job.jobID.asString,
-                job.profile.queueName,
-                job.priorityAsString
-                ]:
-        if search in t.lower():
-          saw_text = True
-          break
-      if not saw_text:
-        return False
-    return True
-
-  return filter(predicate, jobs)
+## All Unused below
+
+# DEAD?
+def dock_jobs(request):
+  username = request.user.username
+  matching_jobs = get_job_count_by_state(request, username)
+  return render("jobs_dock_info.mako", request, {
+    'jobs':matching_jobs
+  }, force_template=True)
+register_status_bar_view(dock_jobs)
 
-##################################
-## Task trackers
 
 def get_tasktrackers(request):
   """
@@ -507,9 +450,6 @@ def get_tasktrackers(request):
   return [ Tracker(tracker) for tracker in request.jt.all_task_trackers().trackers]
 
 
-##################################
-## Jobs
-
 def get_single_job(request, jobid):
   """
   Returns the job which matches jobid.
@@ -517,46 +457,6 @@ def get_single_job(request, jobid):
   return Job.from_id(jt=request.jt, jobid=jobid)
 
 
-def get_matching_jobs(request, check_permission=False, **kwargs):
-  """
-  Returns an array of jobs where the returned
-  jobs are matched by the provided filter arguments.
-
-  If a filter argument is in kwargs it will supersede the same argument
-  in the request object.
-
-  Filter arguments may be jobid, pools, user, tasks, text and state.
-
-  Filter by user ownership if check_permission is set to true.
-  """
-  jobfunc = {"completed" : (request.jt.completed_jobs, ThriftJobState.SUCCEEDED),
-             # Succeeded and completed are synonyms here.
-             "succeeded" : (request.jt.completed_jobs, ThriftJobState.SUCCEEDED),
-             "running" : (request.jt.running_jobs, ThriftJobState.RUNNING),
-             "failed" : (request.jt.failed_jobs, ThriftJobState.FAILED),
-             "killed" : (request.jt.killed_jobs, ThriftJobState.KILLED),
-             "all" : (request.jt.all_jobs, None)}
-  if 'state' in kwargs:
-    selection = kwargs['state']
-  else:
-    selection = request.GET.get("state", "all")
-
-  if 'retired' in kwargs:
-    retired_arg = kwargs['retired']
-  else:
-    retired_arg = request.GET.get("retired", None)
-
-  retired = coerce_bool(retired_arg)
-
-  joblist = jobfunc[selection][0]().jobs
-
-  if retired == True:
-    joblist += request.jt.retired_jobs(jobfunc[selection][1]).jobs
-
-  return [Job.from_thriftjob(request.jt, j)
-          for j in _filter_jobs_by_req(joblist, request, **kwargs)
-          if not check_permission or request.user.is_superuser or j.profile.user == request.user.username]
-
 def get_job_count_by_state(request, username):
   """
   Returns the number of comlpeted, running, and failed jobs for a user.
@@ -578,9 +478,6 @@ def get_job_count_by_state(request, username):
   return res
 
 
-##################################
-## JobBrowser views
-
 def jobbrowser(request):
   """
   jobbrowser.jsp - a - like.
@@ -590,7 +487,7 @@ def jobbrowser(request):
     return lambda job: job.status == state
 
   status = request.jt.cluster_status()
-  alljobs = get_matching_jobs(request)
+  alljobs = [] #get_matching_jobs(request)
   runningjobs = filter(check_job_state('RUNNING'), alljobs)
   completedjobs = filter(check_job_state('COMPLETED'), alljobs)
   failedjobs = filter(check_job_state('FAILED'), alljobs)

+ 206 - 0
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import re
+import time
+
+from desktop.lib.view_util import format_duration_in_millis
+
+from jobbrowser.models import format_unixtime_ms
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class Application:
+
+  def __init__(self, attrs):
+    for attr in attrs.keys():
+      setattr(self, attr, attrs[attr])
+
+    self._fixup()
+
+  def _fixup(self):
+    self.is_mr2 = True
+    jobid = self.id
+    if self.state in ('FINISHED', 'FAILED', 'KILLED'):
+      # When a job is finished
+      jobid = jobid.replace('application', 'job')
+      setattr(self, 'status', self.finalStatus)
+    else:
+      jobid = jobid.replace('job', 'application')
+      setattr(self, 'status', self.state)
+    setattr(self, 'jobId', jobid)
+    setattr(self, 'jobId_short', re.sub('(application|job)_', '', self.jobId))
+    setattr(self, 'jobName', self.name)
+    setattr(self, 'is_retired', False)
+    setattr(self, 'maps_percent_complete', self.progress)
+    setattr(self, 'reduces_percent_complete', self.progress)
+    setattr(self, 'queueName', self.queue)
+    setattr(self, 'priority', '')
+    if self.finishedTime == 0:
+      finishTime = int(time.time() * 1000)
+    else:
+      finishTime = self.finishedTime
+    setattr(self, 'durationInMillis', finishTime - self.startedTime)
+    setattr(self, 'startTimeMs', self.startedTime)
+    setattr(self, 'startTimeFormatted',  format_unixtime_ms(self.startedTime))
+    setattr(self, 'finishedMaps', None)
+    setattr(self, 'desiredMaps', None)
+    setattr(self, 'finishedReduces', None)
+    setattr(self, 'desiredReduces', None)
+    setattr(self, 'durationFormatted', format_duration_in_millis(self.durationInMillis))
+
+
+class Job:
+
+  def __init__(self, api, attrs):
+    self.api = api
+    self.is_mr2 = True
+    for attr in attrs.keys():
+      setattr(self, attr, attrs[attr])
+
+    self._fixup()
+
+  def _fixup(self):
+    jobid = self.id
+
+    if self.state in ('SUCCEEDED', 'FAILED', 'KILL_WAIT', 'KILLED', 'ERROR'):
+      setattr(self, 'status', self.state)
+      # When a job is finished, just use 'job' instead of 'application'
+      jobid = jobid.replace('application', 'job')
+    else:
+      jobid = jobid.replace('job', 'application')
+      setattr(self, 'status', self.state)
+    setattr(self, 'jobId', jobid)
+    setattr(self, 'jobId_short', self.jobId.replace('job_', ''))
+    setattr(self, 'is_retired', True)
+    setattr(self, 'maps_percent_complete', None)
+    setattr(self, 'reduces_percent_complete', None)
+    setattr(self, 'duration', self.finishTime - self.startTime)
+    setattr(self, 'finishTimeFormatted', format_unixtime_ms(self.finishTime))
+    setattr(self, 'startTimeFormatted', format_unixtime_ms(self.startTime))
+    setattr(self, 'finishedMaps', self.mapsCompleted)
+    setattr(self, 'desiredMaps', None)
+    setattr(self, 'finishedReduces', self.reducesCompleted)
+    setattr(self, 'desiredReduces', None)
+
+  @property
+  def counters(self):
+    return self.api.counters(self.id)['jobCounters']
+
+  @property
+  def full_job_conf(self):
+    if not hasattr(self, '_full_job_conf'):
+      self._full_job_conf = self.api.conf(self.id)['conf']
+    return self._full_job_conf
+
+  @property
+  def conf_keys(self):
+    return dict([(line['name'], line['value']) for line in self.full_job_conf['property']])
+
+  def get_task(self, task_id):
+    json = self.api.task(self.id, task_id)['task']
+    return Task(self, json)
+
+  def filter_tasks(self, task_types=None, task_states=None, task_text=None):
+    return [Task(self, task) for task in self.api.tasks(self.id).get('tasks', {}).get('task', [])
+            if (not task_types or task['type'].lower() in task_types) and
+               (not task_states or task['state'].lower() in task_states)]
+
+  @property
+  def job_attempts(self):
+    if not hasattr(self, '_job_attempts'):
+      self._job_attempts = self.api.job_attempts(self.id)['jobAttempts']
+    return self._job_attempts
+
+
+class Task:
+
+  def __init__(self, job, attrs):
+    self.job = job
+    if attrs:
+      for key, value in attrs.iteritems():
+        setattr(self, key, value)
+    self.is_mr2 = True
+
+    self._fixup()
+
+  def _fixup(self):
+    setattr(self, 'jobId', self.job.jobId)
+    setattr(self, 'taskId', self.id)
+    setattr(self, 'taskId_short', self.id)
+    setattr(self, 'taskType', self.type)
+    setattr(self, 'execStartTimeMs', self.startTime)
+    setattr(self, 'mostRecentState', self.state)
+    setattr(self, 'execStartTimeFormatted', format_unixtime_ms(self.startTime))
+    setattr(self, 'execFinishTimeFormatted', format_unixtime_ms(self.finishTime))
+    setattr(self, 'startTimeFormatted', self.startTime)
+
+  @property
+  def attempts(self):
+    # We can cache as we deal with history server
+    if not hasattr(self, '_attempts'):
+      self._attempts = [Attempt(self, attempt) for attempt in self.job.api.task_attempts(self.job.id, self.id)['taskAttempts']['taskAttempt']]
+    return self._attempts
+
+  @property
+  def taskAttemptIds(self):
+    if not hasattr(self, '_taskAttemptIds'):
+      self._taskAttemptIds = [attempt.id for attempt in self.attempts]
+    return self._taskAttemptIds
+
+  @property
+  def counters(self):
+    if not hasattr(self, '_counters'):
+      self._counters = self.job.api.task_counters(self.jobId, self.id)['jobTaskCounters']
+    return self._counters
+
+  def get_attempt(self, attempt_id):
+    json = self.job.api.task_attempt(self.jobId, self.id, attempt_id)['taskAttempt']
+    return Attempt(self, json)
+
+
+class Attempt:
+
+  def __init__(self, task, attrs):
+    self.task = task
+    if attrs:
+      for key, value in attrs.iteritems():
+        setattr(self, key, value)
+    self.is_mr2 = True
+
+    self._fixup()
+
+  def _fixup(self):
+    setattr(self, 'attemptId', self.id)
+    setattr(self, 'attemptId_short', self.id)
+    setattr(self, 'taskTrackerId', self.assignedContainerId)
+    setattr(self, 'startTimeFormatted', self.startTime)
+    setattr(self, 'finishTimeFormatted', self.finishTime)
+    setattr(self, 'outputSize', None)
+    setattr(self, 'phase', None)
+    setattr(self, 'shuffleFinishTimeFormatted', None)
+    setattr(self, 'sortFinishTimeFormatted', None)
+    setattr(self, 'mapFinishTimeFormatted', None)
+
+  @property
+  def counters(self):
+    if not hasattr(self, '_counters'):
+      self._counters = self.task.job.api.task_attempt_counters(self.task.jobId, self.task.id, self.id)['jobCounters']
+    return self._counters

+ 1 - 1
apps/jobsub/src/jobsub/templates/workflow.mako

@@ -128,7 +128,7 @@ ${layout.menubar(section='history')}
 
                 <td>
                 % if action.externalId:
-                  <a href="${ url('jobbrowser.views.single_job', jobid=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
+                  <a href="${ url('jobbrowser.views.single_job', job=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
                 % endif
                 </td>
 

+ 2 - 2
apps/oozie/src/oozie/templates/dashboard/list_oozie_workflow.mako

@@ -200,7 +200,7 @@ ${ layout.menubar(section='dashboard') }
               <tr>
                 <td>
                   % if action.externalId:
-                    <a href="${ url('jobbrowser.views.job_single_logs', jobid=action.externalId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
+                    <a href="${ url('jobbrowser.views.job_single_logs', job=action.externalId) }" data-row-selector-exclude="true"><i class="icon-tasks"></i></a>
                   % endif
                 </td>
                 <td>
@@ -217,7 +217,7 @@ ${ layout.menubar(section='dashboard') }
                 <td><span class="label ${ utils.get_status(action.status) }">${ action.status }</span></td>
                 <td>
                   % if action.externalId:
-                    <a href="${ url('jobbrowser.views.single_job', jobid=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
+                    <a href="${ url('jobbrowser.views.single_job', job=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
                   % endif
                 </td>
 

+ 1 - 1
apps/oozie/src/oozie/templates/dashboard/list_oozie_workflow_action.mako

@@ -87,7 +87,7 @@ ${ layout.menubar(section='running') }
         <td>${ _('External Id') }</td>
         <td>
           % if action.externalId:
-            <a href="${ url('jobbrowser.views.single_job', jobid=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
+            <a href="${ url('jobbrowser.views.single_job', job=action.externalId) }">${ "_".join(action.externalId.split("_")[-2:]) }</a>
           % endif
         </td>
       </tr>

+ 1 - 1
apps/oozie/src/oozie/templates/editor/gen/workflow-graph-status.xml.mako

@@ -54,7 +54,7 @@
         </div>
         <div class="span2">
           % if action and action.externalId:
-           <a href="${ url('jobbrowser.views.job_single_logs', jobid=action.externalId) }" data-row-selector-exclude="true" id="advanced-btn">
+           <a href="${ url('jobbrowser.views.job_single_logs', job=action.externalId) }" data-row-selector-exclude="true" id="advanced-btn">
               <i class="icon-tasks"></i> ${ _('View the logs') }
             </a>
           % endif

+ 9 - 0
desktop/conf.dist/hue.ini

@@ -288,6 +288,15 @@
       # Defaults to $HADOOP_CONF_DIR or /etc/hadoop/conf
       ## hadoop_conf_dir=/etc/hadoop/conf
 
+      # URL of the ResourceManager API
+      ## resourcemanager_api_url=http://localhost:8088
+
+      # URL of the ProxyServer API
+      ## proxy_api_url=http://localhost:8088
+
+      # URL of the HistoryServer API
+      history_server_api_url=http://localhost:19888
+
 
 ###########################################################################
 # Settings to configure liboozie

+ 9 - 0
desktop/conf/pseudo-distributed.ini.tmpl

@@ -291,6 +291,15 @@
       # Defaults to $HADOOP_CONF_DIR or /etc/hadoop/conf
       ## hadoop_conf_dir=/etc/hadoop/conf
 
+      # URL of the ResourceManager API
+      ## resourcemanager_api_url=http://localhost:8088
+
+      # URL of the ProxyServer API
+      ## proxy_api_url=http://localhost:8088
+
+      # URL of the HistoryServer API
+      history_server_api_url=http://localhost:19888
+
 
 ###########################################################################
 # Settings to configure liboozie

+ 2 - 2
desktop/core/src/desktop/lib/rest/resource.py

@@ -80,7 +80,7 @@ class Resource(object):
       return body
 
 
-  def get(self, relpath=None, params=None):
+  def get(self, relpath=None, params=None, headers=None):
     """
     Invoke the GET method on a resource.
     @param relpath: Optional. A relative path to this resource's path.
@@ -88,7 +88,7 @@ class Resource(object):
 
     @return: A dictionary of the JSON result.
     """
-    return self.invoke("GET", relpath, params)
+    return self.invoke("GET", relpath, params, headers=headers)
 
 
   def delete(self, relpath=None, params=None):

+ 3 - 0
desktop/libs/hadoop/src/hadoop/cluster.py

@@ -116,6 +116,9 @@ def get_cluster_addr_for_job_submission():
     return None
   return "%s:%s" % (conf.HOST.get(), conf.PORT.get())
 
+def is_yarn():
+  cluster = get_cluster_conf_for_job_submission()
+  return cluster is not None and 'IS_YARN' in cluster.config.members
 
 def clear_caches():
   """

+ 11 - 0
desktop/libs/hadoop/src/hadoop/conf.py

@@ -208,6 +208,17 @@ YARN_CLUSTERS = UnspecifiedConfigSection(
               "as the --config flag. Defaults to the environment variable " +
               "HADOOP_CONF_DIR when set, or '/etc/hadoop/conf'.")
       ),
+      IS_YARN=Config("is_yarn", help="Attribute set only on YARN clusters and not MR1 ones.",
+                     default=True, type=coerce_bool),
+      RESOURCE_MANAGER_API_URL=Config("resourcemanager_api_url",
+                  default='http://localhost:8088',
+                  help="URL of the ResourceManager API"),
+      PROXY_API_URL=Config("proxy_api_url",
+                  default='http://localhost:8088',
+                  help="URL of the ProxyServer API"),
+      HISTORY_SERVER_API_URL=Config("history_server_api_url",
+                  default='http://localhost:19888',
+                  help="URL of the HistoryServer API"),
     )
   )
 )

+ 16 - 0
desktop/libs/hadoop/src/hadoop/yarn/__init__.py

@@ -0,0 +1,16 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

+ 93 - 0
desktop/libs/hadoop/src/hadoop/yarn/history_server_api.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import posixpath
+import threading
+
+from desktop.lib.rest.http_client import HttpClient
+from desktop.lib.rest.resource import Resource
+from hadoop import cluster
+
+
+LOG = logging.getLogger(__name__)
+DEFAULT_USER = 'hue'
+
+_API_VERSION = 'v1'
+_JSON_CONTENT_TYPE = 'application/json'
+
+_api_cache = None
+_api_cache_lock = threading.Lock()
+
+
+def get_history_server_api():
+  global _api_cache
+  if _api_cache is None:
+    _api_cache_lock.acquire()
+    try:
+      if _api_cache is None:
+        yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+        _api_cache = HistoryServerApi(yarn_cluster.HISTORY_SERVER_API_URL.get())
+    finally:
+      _api_cache_lock.release()
+  return _api_cache
+
+
+class HistoryServerApi(object):
+
+  def __init__(self, oozie_url):
+    self._url = posixpath.join(oozie_url, 'ws/%s/history' % _API_VERSION)
+    self._client = HttpClient(self._url, logger=LOG)
+    self._root = Resource(self._client)
+
+  def __str__(self):
+    return "HistoryServerApi at %s" % (self._url,)
+
+  @property
+  def url(self):
+    return self._url
+
+  def job(self, user, job_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s' % {'job_id': job_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def counters(self, job_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/counters' % {'job_id': job_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def conf(self, job_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/conf' % {'job_id': job_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def job_attempts(self, job_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/jobattempts' % {'job_id': job_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def tasks(self, job_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks' % {'job_id': job_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task(self, job_id, task_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks/%(task_id)s' % {'job_id': job_id, 'task_id': task_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_attempts(self, job_id, task_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/attempts' % {'job_id': job_id, 'task_id': task_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_counters(self, job_id, task_id):
+    job_id = job_id.replace('application', 'job')
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/counters' % {'job_id': job_id, 'task_id': task_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_attempt(self, job_id, task_id, attempt_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/attempts/%(attempt_id)s' % {'job_id': job_id, 'task_id': task_id, 'attempt_id': attempt_id}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_attempt_counters(self, job_id, task_id, attempt_id):
+    return self._root.get('mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/attempts/%(attempt_id)s/counters' % {'job_id': job_id, 'task_id': task_id, 'attempt_id': attempt_id}, headers={'Accept': _JSON_CONTENT_TYPE})

+ 97 - 0
desktop/libs/hadoop/src/hadoop/yarn/mapreduce_api.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import posixpath
+import threading
+
+from desktop.lib.rest.http_client import HttpClient
+from desktop.lib.rest.resource import Resource
+from hadoop import cluster
+
+
+LOG = logging.getLogger(__name__)
+DEFAULT_USER = 'hue'
+
+_API_VERSION = 'v1'
+_JSON_CONTENT_TYPE = 'application/json'
+
+_api_cache = None
+_api_cache_lock = threading.Lock()
+
+
+def get_mapreduce_api():
+  global _api_cache
+  if _api_cache is None:
+    _api_cache_lock.acquire()
+    try:
+      if _api_cache is None:
+        yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+        _api_cache = MapreduceApi(yarn_cluster.PROXY_API_URL.get())
+    finally:
+      _api_cache_lock.release()
+  return _api_cache
+
+
+class MapreduceApi(object):
+
+  def __init__(self, oozie_url):
+    self._url = posixpath.join(oozie_url, 'proxy')
+    self._client = HttpClient(self._url, logger=LOG)
+    self._root = Resource(self._client)
+    self._security_enabled = False
+
+  def __str__(self):
+    return "MapreduceApi at %s" % (self._url,)
+
+  @property
+  def url(self):
+    return self._url
+
+  def job(self, user, job_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s' % {'app_id': app_id, 'job_id': job_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def counters(self, job_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/counters' % {'app_id': app_id, 'job_id': job_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def tasks(self, job_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/tasks' % {'app_id': app_id, 'job_id': job_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def job_attempts(self, job_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/jobattempts' % {'app_id': app_id, 'job_id': job_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def conf(self, job_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/conf' % {'app_id': app_id, 'job_id': job_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task(self, job_id, task_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/tasks/%(task_id)s' % {'app_id': app_id, 'job_id': job_id, 'task_id': task_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_counters(self, job_id, task_id):
+    app_id = job_id.replace('job', 'application')
+    job_id = job_id.replace('application', 'job')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/counters' % {'app_id': app_id, 'job_id': job_id, 'task_id': task_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def task_attempts(self, job_id, task_id):
+    app_id = job_id.replace('job', 'application')
+    return self._root.get('%(app_id)s/ws/%(version)s/mapreduce/jobs/%(job_id)s/tasks/%(task_id)s/attempts' % {'app_id': app_id, 'job_id': job_id, 'task_id': task_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})
+

+ 69 - 0
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import posixpath
+import threading
+
+from desktop.lib.rest.http_client import HttpClient
+from desktop.lib.rest.resource import Resource
+from hadoop import cluster
+
+
+LOG = logging.getLogger(__name__)
+DEFAULT_USER = 'hue'
+
+_API_VERSION = 'v1'
+_JSON_CONTENT_TYPE = 'application/json'
+
+_api_cache = None
+_api_cache_lock = threading.Lock()
+
+
+def get_resource_manager():
+  global _api_cache
+  if _api_cache is None:
+    _api_cache_lock.acquire()
+    try:
+      if _api_cache is None:
+        yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+        _api_cache = ResourceManagerApi(yarn_cluster.RESOURCE_MANAGER_API_URL.get())
+    finally:
+      _api_cache_lock.release()
+  return _api_cache
+
+
+class ResourceManagerApi(object):
+  def __init__(self, oozie_url):
+    self._url = posixpath.join(oozie_url, 'ws', _API_VERSION)
+    self._client = HttpClient(self._url, logger=LOG)
+    self._root = Resource(self._client)
+    self._security_enabled = False
+
+  def __str__(self):
+    return "ResourceManagerApi at %s" % (self._url,)
+
+  @property
+  def url(self):
+    return self._url
+
+  @property
+  def security_enabled(self):
+    return self._security_enabled
+
+  def apps(self, **kwargs):
+    return self._root.get('cluster/apps', params=kwargs, headers={'Accept': _JSON_CONTENT_TYPE})