Ver código fonte

HUE-7945 [jb] Add oozie launcher job browser type

jdesjean 7 anos atrás
pai
commit
0219bfc

+ 6 - 2
apps/jobbrowser/src/jobbrowser/api.py

@@ -33,7 +33,7 @@ import hadoop.yarn.resource_manager_api as resource_manager_api
 import hadoop.yarn.spark_history_server_api as spark_history_server_api
 
 from jobbrowser.conf import SHARE_JOBS
-from jobbrowser.yarn_models import Application, Job as YarnJob, KilledJob as KilledYarnJob, Container, SparkJob
+from jobbrowser.yarn_models import Application, OozieYarnJob, Job as YarnJob, KilledJob as KilledYarnJob, Container, SparkJob
 
 
 LOG = logging.getLogger(__name__)
@@ -149,10 +149,14 @@ class YarnApi(JobBrowserApi):
           job = SparkJob(app, rm_api=self.resource_manager_api, hs_api=self.spark_history_server_api)
         elif app['state'] in ('KILLED', 'FAILED'):
           job = KilledYarnJob(self.resource_manager_api, app)
+        elif app['applicationType'] == 'Oozie Launcher':
+          job = OozieYarnJob(self.resource_manager_api, app)
         else:  # Job succeeded, attempt to fetch from JHS
           job = self._get_job_from_history_server(job_id)
       else:
-        if app['state'] == 'ACCEPTED':
+        if app['applicationType'] == 'Oozie Launcher':
+          job = OozieYarnJob(self.resource_manager_api, app)
+        elif app['state'] == 'ACCEPTED':
           raise ApplicationNotRunning(app_id, app)
         # The MapReduce API only returns JSON when the application is in a RUNNING state
         elif app['state'] in ('NEW', 'SUBMITTED', 'RUNNING') and app['applicationType'] == 'MAPREDUCE':

+ 35 - 18
apps/jobbrowser/src/jobbrowser/apis/job_api.py

@@ -67,7 +67,7 @@ class JobApi(Api):
       return self.yarn_api
     elif appid.startswith('task_'):
       return YarnMapReduceTaskApi(self.user, appid)
-    elif appid.startswith('attempt_'):
+    elif appid.startswith('attempt_') or appid.startswith('appattempt_'):
       return YarnMapReduceTaskAttemptApi(self.user, appid)
     elif appid.find('_executor_') > 0:
       return SparkExecutorApi(self.user, appid)
@@ -147,7 +147,7 @@ class YarnApi(Api):
         'name': app['name'],
         'type': app['applicationType'],
         'status': app['status'],
-        'apiStatus': self._api_status(app['status']),
+        'apiStatus': self._api_status(app['status'], app['applicationType']),
         'user': app['user'],
         'progress': app['progress'],
         'duration': app['durationMs'],
@@ -186,6 +186,13 @@ class YarnApi(Api):
       }
       if hasattr(job, 'metrics'):
         common['metrics'] = job.metrics
+    elif app['applicationType'] == 'Oozie Launcher':
+      common['properties'] = {
+        'startTime': job.startTime,
+        'finishTime': job.finishTime,
+        'elapsedTime': job.duration,
+        'attempts': []
+      }
 
     return common
 
@@ -206,7 +213,7 @@ class YarnApi(Api):
   def logs(self, appid, app_type, log_name, is_embeddable=False):
     logs = ''
     try:
-      if app_type == 'MAPREDUCE':
+      if app_type == 'MAPREDUCE' or app_type == 'Oozie Launcher':
         if log_name == 'default':
           response = job_single_logs(MockDjangoRequest(self.user), job=appid)
           logs = json.loads(response.content).get('logs')
@@ -242,12 +249,18 @@ class YarnApi(Api):
           'executor_list': NativeYarnApi(self.user).get_job(jobid=appid).get_executors(),
           'filter_text': ''
         }
+    elif app_type == 'Oozie Launcher':
+      if app_property == 'attempts':
+        return {
+          'task_list': NativeYarnApi(self.user).get_job(jobid=appid).job_attempts['jobAttempt'],
+          'filter_text': ''
+        }
     return {}
 
-  def _api_status(self, status):
+  def _api_status(self, status, app_type=None):
     if status in ['NEW', 'NEW_SAVING', 'SUBMITTED', 'ACCEPTED', 'RUNNING']:
       return 'RUNNING'
-    elif status == 'SUCCEEDED':
+    elif status == 'SUCCEEDED' or (app_type == 'Oozie Launcher' and status == 'FINISHED'):
       return 'SUCCEEDED'
     else:
       return 'FAILED' # FAILED, KILLED
@@ -359,8 +372,9 @@ class YarnMapReduceTaskAttemptApi(Api):
 
   def __init__(self, user, app_id):
     Api.__init__(self, user)
-    self.app_id = '_'.join(app_id.replace('task_', 'application_').replace('attempt_', 'application_').split('_')[:3])
-    self.task_id = '_'.join(app_id.replace('attempt_', 'task_').split('_')[:5])
+    start = 'appattempt_' if app_id.startswith('appattempt_') else 'attempt_'
+    self.app_id = '_'.join(app_id.replace('task_', 'application_').replace(start, 'application_').split('_')[:3])
+    self.task_id = '_'.join(app_id.replace(start, 'task_').split('_')[:5])
     self.attempt_id = app_id
 
 
@@ -416,22 +430,25 @@ class YarnMapReduceTaskAttemptApi(Api):
     return {
         #"elapsedMergeTime" : task.elapsedMergeTime,
         #"shuffleFinishTime" : task.shuffleFinishTime,
-        "assignedContainerId" : task.assignedContainerId,
-        "progress" : task.progress,
-        "elapsedTime" : task.elapsedTime,
-        "state" : task.state,
+        "assignedContainerId" : task.assignedContainerId if hasattr(task, 'assignedContainerId') else task.amContainerId if hasattr(task, 'amContainerId') else '',
+        "progress" : task.progress if hasattr(task, 'progress') else '',
+        "elapsedTime" : task.elapsedTime if hasattr(task, 'elapsedTime') else '',
+        "state" : task.state if hasattr(task, 'state') else task.appAttemptState if hasattr(task, 'appAttemptState') else '',
         #"elapsedShuffleTime" : task.elapsedShuffleTime,
         #"mergeFinishTime" : task.mergeFinishTime,
-        "rack" : task.rack,
+        "rack" : task.rack if hasattr(task, 'rack') else '',
         #"elapsedReduceTime" : task.elapsedReduceTime,
-        "nodeHttpAddress" : task.nodeHttpAddress,
-        "type" : task.type + '_ATTEMPT',
-        "startTime" : task.startTime,
-        "id" : task.id,
-        "finishTime" : task.finishTime,
+        "nodeHttpAddress" : task.nodeHttpAddress if hasattr(task, 'nodeHttpAddress') else '',
+        "type" : task.type + '_ATTEMPT' if hasattr(task, 'type') else '',
+        "startTime" : task.startTime if hasattr(task, 'startTime') else '',
+        "id" : task.id if hasattr(task, 'id') else task.appAttemptId if hasattr(task, 'appAttemptId') else '',
+        "finishTime" : task.finishTime if hasattr(task, 'finishTime') else long(task.finishedTime) if hasattr(task, 'finishedTime') else '',
         "app_id": self.app_id,
         "task_id": self.task_id,
-        'apiStatus': self._api_status(task.state),
+        'apiStatus': self._api_status(task.state) if hasattr(task, 'state') else self._api_status(task.appAttemptState) if hasattr(task, 'appAttemptState') else '',
+        'host': task.host if hasattr(task, 'host') else '',
+        'rpcPort': task.rpcPort if hasattr(task, 'rpcPort') else '',
+        'diagnosticsInfo': task.diagnosticsInfo if hasattr(task, 'diagnosticsInfo') else ''
     }
 
 

+ 124 - 33
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -418,8 +418,12 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
     <div data-bind="template: { name: 'job-yarn-page${ SUFFIX }', data: $root.job() }"></div>
   <!-- /ko -->
 
-  <!-- ko if: type() == 'IMPALA' -->
-    <div data-bind="template: { name: 'job-impala-page${ SUFFIX }', data: $root.job() }"></div>
+  <!-- ko if: type() == 'Oozie Launcher' -->
+    <div data-bind="template: { name: 'job-oozie-page${ SUFFIX }', data: $root.job() }"></div>
+  <!-- /ko -->
+
+  <!-- ko if: type() == 'Oozie Launcher_ATTEMPT' -->
+    <div data-bind="template: { name: 'job-oozie-attempt-page${ SUFFIX }', data: $root.job() }"></div>
   <!-- /ko -->
 
   <!-- ko if: type() == 'SPARK' -->
@@ -687,8 +691,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
 </script>
 
-
-<script type="text/html" id="job-mapreduce-task-attempt-page${ SUFFIX }">
+<script type="text/html" id="job-oozie-page${ SUFFIX }">
 
   <div class="row-fluid">
     <div data-bind="css:{'span2': !$root.isMini(), 'span12': $root.isMini() }">
@@ -706,15 +709,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             </div>
           </li>
           <!-- ko if: !$root.isMini() -->
+          <li class="nav-header">${ _('State') }</li>
+          <li><span data-bind="text: status"></span></li>
           <!-- ko with: properties -->
-            <li class="nav-header">${ _('State') }</li>
-            <li><span data-bind="text: state"></span></li>
-            <li class="nav-header">${ _('Assigned Container ID') }</li>
-            <li><span data-bind="text: assignedContainerId"></span></li>
-            <li class="nav-header">${ _('Rack') }</li>
-            <li><span data-bind="text: rack"></span></li>
-            <li class="nav-header">${ _('Node HTTP address') }</li>
-            <li><span data-bind="text: nodeHttpAddress"></span></li>
             <li class="nav-header">${ _('Start time') }</li>
             <li><span data-bind="moment: {data: startTime, format: 'LLL'}"></span></li>
             <li class="nav-header">${ _('Finish time') }</li>
@@ -730,23 +727,49 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
     <div data-bind="css: {'span10': !$root.isMini(), 'span12': $root.isMini() }">
 
       <ul class="nav nav-pills margin-top-20">
-        <li class="active"><a class="jb-logs-link" href="#job-mapreduce-task-attempt-page-logs${ SUFFIX }" data-toggle="tab">${ _('Logs') }</a></li>
-        <li><a href="#job-mapreduce-task-attempt-page-counters${ SUFFIX }" data-bind="click: function(){ fetchProfile('counters'); $('a[href=\'#job-mapreduce-task-attempt-page-counters${ SUFFIX }\']').tab('show'); }">${ _('Counters') }</a></li>
+        <li class="active"><a class="jb-logs-link" href="#job-oozie-page-logs${ SUFFIX }" data-toggle="tab">${ _('Logs') }</a></li>
+        <li><a href="#job-oozie-page-attempts${ SUFFIX }" data-bind="click: function(){ fetchProfile('attempts'); $('a[href=\'#job-oozie-page-attempts${ SUFFIX }\']').tab('show'); }">${ _('Attempts') }</a></li>
       </ul>
 
       <div class="tab-content">
-        <div class="tab-pane active" id="job-mapreduce-task-attempt-page-logs${ SUFFIX }">
+        <div class="tab-pane active" id="job-oozie-page-logs${ SUFFIX }">
           <ul class="nav nav-tabs">
-          % for name in ['stdout', 'stderr', 'syslog']:
+          % for name in ['stdout', 'stderr']:
             <li class="${ name == 'stdout' and 'active' or '' }"><a href="javascript:void(0)" data-bind="click: function(data, e) { $(e.currentTarget).parent().siblings().removeClass('active'); $(e.currentTarget).parent().addClass('active'); fetchLogs('${ name }'); }, text: '${ name }'"></a></li>
           % endfor
           </ul>
           <pre data-bind="html: logs, logScroller: logs"></pre>
         </div>
 
-        <div class="tab-pane" id="job-mapreduce-task-attempt-page-counters${ SUFFIX }">
-          <div data-bind="template: { name: 'render-attempt-counters${ SUFFIX }', data: properties['counters'] }"></div>
+        <div class="tab-pane" id="job-oozie-page-attempts${ SUFFIX }">
+          <table class="table table-condensed">
+            <thead>
+            <tr>
+              <th>${_('Assigned Container Id')}</th>
+              <th>${_('Node Id')}</th>
+              <th>${_('appAttemptId')}</th>
+              <th>${_('Start Time')}</th>
+              <th>${_('Finish Time')}</th>
+              <th>${_('Node Http Address')}</th>
+              <th>${_('Blacklisted Nodes')}</th>
+              <th>${_('Nodes Blacklisted By System')}</th>
+            </tr>
+            </thead>
+            <tbody data-bind="foreach: properties['attempts']()['task_list']">
+              <tr class="pointer" data-bind="click: function() { $root.job().id(id); $root.job().fetchJob(); }">
+                <td data-bind="text: containerId"></td>
+                <td data-bind="text: nodeId"></td>
+                <td data-bind="text: appAttemptId"></td>
+                <td data-bind="moment: {data: startTime, format: 'LLL'}"></td>
+                <td data-bind="moment: {data: finishedTime, format: 'LLL'}"></td>
+                <td data-bind="text: nodeHttpAddress"></td>
+                <td data-bind="text: blacklistedNodes"></td>
+                <td data-bind="text: nodesBlacklistedBySystem"></td>
+              </tr>
+            </tbody>
+          </table>
         </div>
+
       </div>
 
     </div>
@@ -754,9 +777,45 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
 </script>
 
+<script type="text/html" id="job-oozie-attempt-page${ SUFFIX }">
 
-<script type="text/html" id="job-impala-page${ SUFFIX }">
-   <div class="row-fluid">
+  <div class="row-fluid">
+    <div data-bind="css:{'span2': !$root.isMini(), 'span12': $root.isMini() }">
+      <div class="sidebar-nav">
+        <ul class="nav nav-list">
+          <li class="nav-header">${ _('Id') }</li>
+          <li class="break-word"><span data-bind="text: id"></span></li>
+          <li class="nav-header">${ _('Type') }</li>
+          <li><span data-bind="text: type"></span></li>
+          <!-- ko with: properties -->
+          <li class="nav-header">${ _('State') }</li>
+          <li><span data-bind="text: state"></span></li>
+          <!-- ko if: !$root.isMini() -->
+          <li class="nav-header">${ _('Finish time') }</li>
+          <li><span data-bind="moment: {data: finishTime, format: 'LLL'}"></span></li>
+          <li class="nav-header">${ _('Assigned Container ID') }</li>
+          <li><span data-bind="text: assignedContainerId"></span></li>
+          <li class="nav-header">${ _('Host') }</li>
+          <li><span data-bind="text: host"></span></li>
+          <li class="nav-header">${ _('RPC Port') }</li>
+          <li><span data-bind="text: rpcPort"></span></li>
+          <li class="nav-header">${ _('Diagnostics Info') }</li>
+          <li><span data-bind="text: diagnosticsInfo"></span></li>
+          <!-- /ko -->
+          <!-- /ko -->
+        </ul>
+      </div>
+    </div>
+
+    <div data-bind="css: {'span10': !$root.isMini(), 'span12': $root.isMini() }">
+    </div>
+  </div>
+
+</script>
+
+<script type="text/html" id="job-mapreduce-task-attempt-page${ SUFFIX }">
+
+  <div class="row-fluid">
     <div data-bind="css:{'span2': !$root.isMini(), 'span12': $root.isMini() }">
       <div class="sidebar-nav">
         <ul class="nav nav-list">
@@ -764,32 +823,61 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           <li class="break-word"><span data-bind="text: id"></span></li>
           <li class="nav-header">${ _('Type') }</li>
           <li><span data-bind="text: type"></span></li>
-          <li class="nav-header">${ _('Status') }</li>
-          <li><span data-bind="text: status"></span></li>
-          <li class="nav-header">${ _('User') }</li>
-          <li><span data-bind="text: user"></span></li>
           <li class="nav-header">${ _('Progress') }</li>
           <li><span data-bind="text: progress"></span>%</li>
           <li>
-            <div class="progress-job progress" style="background-color: #FFF; width: 100%" data-bind="css: {'progress-warning': apiStatus() !== 'FAILED' && progress() < 100, 'progress-success': apiStatus() !== 'FAILED' && progress() === 100, 'progress-danger': apiStatus() === 'FAILED'}">
+            <div class="progress-job progress" style="background-color: #FFF; width: 100%" data-bind="css: {'progress-warning': apiStatus() == 'RUNNING', 'progress-success': apiStatus() == 'SUCCEEDED', 'progress-danger': apiStatus() == 'FAILED'}, attr: {title: status}">
               <div class="bar" data-bind="style: {'width': progress() + '%'}"></div>
             </div>
           </li>
           <!-- ko if: !$root.isMini() -->
-          <li class="nav-header">${ _('Open Duration') }</li>
-          <li><span data-bind="text: duration().toHHMMSS()"></span></li>
-          <li class="nav-header">${ _('Submitted') }</li>
-          <li><span data-bind="text: submitted"></span></li>
+          <!-- ko with: properties -->
+            <li class="nav-header">${ _('State') }</li>
+            <li><span data-bind="text: state"></span></li>
+            <li class="nav-header">${ _('Assigned Container ID') }</li>
+            <li><span data-bind="text: assignedContainerId"></span></li>
+            <li class="nav-header">${ _('Rack') }</li>
+            <li><span data-bind="text: rack"></span></li>
+            <li class="nav-header">${ _('Node HTTP address') }</li>
+            <li><span data-bind="text: nodeHttpAddress"></span></li>
+            <li class="nav-header">${ _('Start time') }</li>
+            <li><span data-bind="moment: {data: startTime, format: 'LLL'}"></span></li>
+            <li class="nav-header">${ _('Finish time') }</li>
+            <li><span data-bind="moment: {data: finishTime, format: 'LLL'}"></span></li>
+            <li class="nav-header">${ _('Elapsed time') }</li>
+            <li><span data-bind="text: elapsedTime().toHHMMSS()"></span></li>
+          <!-- /ko -->
           <!-- /ko -->
         </ul>
       </div>
     </div>
-    <div data-bind="css:{'span10': !$root.isMini(), 'span12 no-margin': $root.isMini() }">
-      <div class="pull-right" data-bind="template: { name: 'job-actions${ SUFFIX }' }"></div>
+
+    <div data-bind="css: {'span10': !$root.isMini(), 'span12': $root.isMini() }">
+
+      <ul class="nav nav-pills margin-top-20">
+        <li class="active"><a class="jb-logs-link" href="#job-mapreduce-task-attempt-page-logs${ SUFFIX }" data-toggle="tab">${ _('Logs') }</a></li>
+        <li><a href="#job-mapreduce-task-attempt-page-counters${ SUFFIX }" data-bind="click: function(){ fetchProfile('counters'); $('a[href=\'#job-mapreduce-task-attempt-page-counters${ SUFFIX }\']').tab('show'); }">${ _('Counters') }</a></li>
+      </ul>
+
+      <div class="tab-content">
+        <div class="tab-pane active" id="job-mapreduce-task-attempt-page-logs${ SUFFIX }">
+          <ul class="nav nav-tabs">
+          % for name in ['stdout', 'stderr', 'syslog']:
+            <li class="${ name == 'stdout' and 'active' or '' }"><a href="javascript:void(0)" data-bind="click: function(data, e) { $(e.currentTarget).parent().siblings().removeClass('active'); $(e.currentTarget).parent().addClass('active'); fetchLogs('${ name }'); }, text: '${ name }'"></a></li>
+          % endfor
+          </ul>
+          <pre data-bind="html: logs, logScroller: logs"></pre>
+        </div>
+
+        <div class="tab-pane" id="job-mapreduce-task-attempt-page-counters${ SUFFIX }">
+          <div data-bind="template: { name: 'render-attempt-counters${ SUFFIX }', data: properties['counters'] }"></div>
+        </div>
+      </div>
+
     </div>
   </div>
-</script>
 
+</script>
 
 <script type="text/html" id="job-spark-page${ SUFFIX }">
    <div class="row-fluid">
@@ -1863,6 +1951,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.canWrite = ko.observableDefault(job.canWrite == true);
 
       self.logs = ko.observable('');
+
       self.properties = ko.mapping.fromJS(job.properties || {});
       self.mainType = ko.observable(vm.interface());
 
@@ -2034,7 +2123,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
               hueUtils.changeURL('#!id=' + vm.job().id());
             }
             var crumbs = [];
-
+            if (/^appattempt_/.test(vm.job().id())) {
+              crumbs.push({'id': vm.job().properties['app_id'], 'name': vm.job().properties['app_id'], 'type': 'app'});
+            }
             if (/^attempt_/.test(vm.job().id())) {
               crumbs.push({'id': vm.job().properties['app_id'], 'name': vm.job().properties['app_id'], 'type': 'app'});
               crumbs.push({'id': vm.job().properties['task_id'], 'name': vm.job().properties['task_id'], 'type': 'task'});

+ 3 - 5
apps/jobbrowser/src/jobbrowser/views.py

@@ -179,7 +179,7 @@ def massage_job_for_json(job, request=None, user=None):
     'user': job.user,
     'isRetired': job.is_retired,
     'isMR2': job.is_mr2,
-    'progress': hasattr(job, 'progress') and job.progress or '',
+    'progress': hasattr(job, 'progress') and job.progress or 0,
     'mapProgress': hasattr(job, 'mapProgress') and job.mapProgress or '',
     'reduceProgress': hasattr(job, 'reduceProgress') and job.reduceProgress or '',
     'setupProgress': hasattr(job, 'setupProgress') and job.setupProgress or '',
@@ -348,6 +348,8 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=L
           log_link = log_link.replace(attempt['nodeHttpAddress'], attempt['nodeId'])
       elif app['state'] == 'RUNNING':
         log_link = app['amContainerLogs']
+    elif app['applicationType'] == 'Oozie Launcher':
+      log_link = app['amContainerLogs']
   except (KeyError, RestException), e:
     raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
   except Exception, e:
@@ -521,10 +523,6 @@ def single_task_attempt_logs(request, job, taskid, attemptid, offset=LOG_OFFSET_
     log_tab = [i for i, log in enumerate(logs) if log]
     if log_tab:
       first_log_tab = log_tab[0]
-  except TaskTrackerNotFoundException:
-    # Four entries,
-    # for diagnostic, stdout, stderr and syslog
-    logs = [_("Failed to retrieve log. TaskTracker not found.")] * 4
   except urllib2.URLError:
     logs = [_("Failed to retrieve log. TaskTracker not ready.")] * 4
 

+ 101 - 4
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -317,6 +317,86 @@ class Job(object):
       self._job_attempts = self.api.job_attempts(self.id)['jobAttempts']
     return self._job_attempts
 
+class OozieYarnJob(Job):
+  def __init__(self, api, attrs):
+    self.api = api
+    for attr in attrs.keys():
+      if attr == 'acls':
+        # 'acls' are actually not available in the API
+        LOG.warn('Not using attribute: %s' % attrs[attr])
+      else:
+        setattr(self, attr, attrs[attr])
+
+    self._fixup()
+
+  def _fixup(self):
+    jobid = self.id
+
+    setattr(self, 'status', self.state)
+    setattr(self, 'jobName', self.name)
+    setattr(self, 'jobId', jobid)
+    setattr(self, 'jobId_short', self.jobId.replace('job_', ''))
+    setattr(self, 'is_retired', False)
+    setattr(self, 'is_mr2', True)
+    setattr(self, 'maps_percent_complete', None)
+    setattr(self, 'reduces_percent_complete', None)
+    setattr(self, 'finishedMaps', 0)
+    setattr(self, 'desiredMaps', 0)
+    setattr(self, 'finishedReduces', 0)
+    setattr(self, 'desiredReduces', 0)
+
+    if self.finishedTime == 0:
+      finishTime = int(time.time() * 1000)
+    else:
+      finishTime = self.finishedTime
+    if self.startedTime == 0:
+      durationInMillis = None
+    else:
+      durationInMillis = finishTime - self.startedTime
+
+    setattr(self, 'duration', durationInMillis)
+    setattr(self, 'durationInMillis', durationInMillis)
+    setattr(self, 'durationFormatted', self.duration and format_duration_in_millis(self.duration))
+    setattr(self, 'finishTimeFormatted', format_unixtime_ms(finishTime))
+    setattr(self, 'startTimeFormatted', format_unixtime_ms(self.startedTime))
+    setattr(self, 'startTimeMs', self.startTimeFormatted)
+    setattr(self, 'startTime', self.startedTime)
+    setattr(self, 'finishTime', finishTime)
+
+    try:
+      setattr(self, 'assignedContainerId', urlparse.urlsplit(self.amContainerLogs).path.split('/node/containerlogs/')[1].split('/')[0])
+    except Exception:
+      setattr(self, 'assignedContainerId', '')
+
+  def get_task(self, task_id):
+    task = YarnTask(self)
+    task.taskId = None
+    task.taskAttemptIds = [appAttempt['appAttemptId'] for appAttempt in self.job_attempts['jobAttempt']]
+    return task
+
+  def filter_tasks(self, task_types=None, task_states=None, task_text=None):
+    return [self.get_task(0)]
+
+  @property
+  def job_attempts(self):
+    if not hasattr(self, '_job_attempts'):
+      attempts = self.api.appattempts(self.id)['appAttempts']['appAttempt']
+      for attempt in attempts:
+        attempt['id'] = attempt['appAttemptId']
+      self._job_attempts = {
+        'jobAttempt': attempts
+      }
+
+    return self._job_attempts
+
+# There's are tasks for Oozie workflow so we create a dummy one.
+class YarnTask:
+  def __init__(self, job):
+    self.job = job
+
+  def get_attempt(self, attempt_id):
+    json = self.job.api.appattempts_attempt(self.job.id, attempt_id)
+    return YarnOozieAttempt(self, json)
 
 class KilledJob(Job):
 
@@ -445,7 +525,7 @@ class Attempt:
     log_link = attempt['logsLink']
 
     # Generate actual task log link from logsLink url
-    if self.task.job.status in ('NEW', 'SUBMITTED', 'RUNNING'):
+    if self.task.job.status in ('NEW', 'SUBMITTED', 'RUNNING') or self.type == 'Oozie Launcher':
       logs_path = '/node/containerlogs/'
       node_url, tracking_path = log_link.split(logs_path)
       container_id, user = tracking_path.strip('/').split('/')
@@ -483,9 +563,13 @@ class Attempt:
 
     for name in ('stdout', 'stderr', 'syslog'):
       link = '/%s/' % name
-      params = {
-        'doAs': user
-      }
+      if self.type == 'Oozie Launcher' and not self.task.job.status == 'FINISHED': # Yarn currently dumps with 500 error with doas in running state
+        params = {}
+      else:
+        params = {
+          'doAs': user
+        }
+
       if int(offset) != 0:
         params['start'] = offset
       else:
@@ -511,6 +595,19 @@ class Attempt:
 
     return logs + [''] * (3 - len(logs))
 
+class YarnOozieAttempt(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, 'diagnostics', self.diagnosticsInfo)
+    setattr(self, 'type', 'Oozie Launcher')
+    setattr(self, 'id', self.appAttemptId)
 
 class Container:
 

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

@@ -128,6 +128,10 @@ class ResourceManagerApi(object):
     params = self._get_params()
     return self._execute(self._root.get, 'cluster/apps/%(app_id)s/appattempts' % {'app_id': app_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
 
+  def appattempts_attempt(self, app_id, attempt_id):
+    params = self._get_params()
+    return self._execute(self._root.get, 'cluster/apps/%(app_id)s/appattempts/%(attempt_id)s' % {'app_id': app_id, 'attempt_id': attempt_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
+
   def kill(self, app_id):
     data = {'state': 'KILLED'}
     token = None