Pārlūkot izejas kodu

HUE-7258 [jb] Properly fetch YARN Spark job logs

Ying Chen 8 gadi atpakaļ
vecāks
revīzija
dc05d52

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

@@ -160,6 +160,8 @@ class YarnApi(JobBrowserApi):
           if not isinstance(resp, dict):
             raise PopupException(_('Mapreduce Proxy API did not return JSON response, check if the job is running.'))
           job = YarnJob(self.mapreduce_api, resp['job'])
+        elif app['state'] in ('NEW', 'SUBMITTED', 'RUNNING') and app['applicationType'] == 'SPARK':
+          job = SparkJob(app, rm_api=self.resource_manager_api, hs_api=self.spark_history_server_api)
         else:
           job = Application(app, self.resource_manager_api)
     except RestException, e:

+ 77 - 5
apps/jobbrowser/src/jobbrowser/apis/job_api.py

@@ -23,11 +23,12 @@ from hadoop.yarn import resource_manager_api
 
 from desktop.lib.exceptions import MessageException
 from desktop.lib.exceptions_renderable import PopupException
-from jobbrowser.conf import MAX_JOB_FETCH
-from jobbrowser.views import job_single_logs
+from jobbrowser.conf import MAX_JOB_FETCH, LOG_OFFSET
+from jobbrowser.views import job_executor_logs, job_single_logs
 
 
 LOG = logging.getLogger(__name__)
+LOG_OFFSET_BYTES = LOG_OFFSET.get()
 
 
 try:
@@ -68,6 +69,8 @@ class JobApi(Api):
       return YarnMapReduceTaskApi(self.user, appid)
     elif appid.startswith('attempt_'):
       return YarnMapReduceTaskAttemptApi(self.user, appid)
+    elif appid.find('_executor_') > 0:
+      return SparkExecutorApi(self.user, appid)
     else:
       return self.yarn_api # application_
 
@@ -170,16 +173,19 @@ class YarnApi(Api):
           'durationFormatted': app['durationFormatted'],
           'startTimeFormatted': app['startTimeFormatted'],
           'diagnostics': app['diagnostics'] if app['diagnostics'] else '',
-
           'tasks': [],
           'metadata': [],
           'counters': []
       }
     elif app['applicationType'] == 'SPARK':
+      app['logs'] = job.logs_url if hasattr(job, 'logs_url') else ''
       common['type'] = 'SPARK'
       common['properties'] = {
-        'metadata': [{'name': name, 'value': value} for name, value in app.iteritems()]
+        'metadata': [{'name': name, 'value': value} for name, value in app.iteritems()],
+        'executors': []
       }
+      if hasattr(job, 'metrics'):
+        common['metrics'] = job.metrics
 
     return common
 
@@ -209,6 +215,9 @@ class YarnApi(Api):
         else:
           response = job_attempt_logs_json(MockDjangoRequest(self.user), job=appid, name=log_name)
           logs = json.loads(response.content).get('log')
+      elif app_type == 'SPARK':
+        response = job_executor_logs(MockDjangoRequest(self.user), job=appid, name=log_name)
+        logs = json.loads(response.content).get('log')
       else:
         logs = None
     except PopupException, e:
@@ -227,7 +236,12 @@ class YarnApi(Api):
         return NativeYarnApi(self.user).get_job(jobid=appid).full_job_conf
       elif app_property == 'counters':
         return NativeYarnApi(self.user).get_job(jobid=appid).counters
-
+    elif app_type == 'SPARK':
+      if app_property == 'executors':
+        return {
+          'executor_list': NativeYarnApi(self.user).get_job(jobid=appid).get_executors(),
+          'filter_text': ''
+        }
     return {}
 
   def _api_status(self, status):
@@ -423,3 +437,61 @@ class YarnMapReduceTaskAttemptApi(Api):
 
 class YarnAtsApi(Api):
   pass
+
+
+class SparkExecutorApi(Api):
+
+  def __init__(self, user, app_id):
+    Api.__init__(self, user)
+    self.app_executor_id = app_id
+    self.executor_id, self.app_id = app_id.split('_executor_')
+    job = NativeYarnApi(self.user).get_job(jobid=self.app_id)
+    if job:
+      executors = job.get_executors()
+      self._executors = [executor for executor in executors if executor['executor_id'] == self.executor_id]
+      self.history_server_api = job.history_server_api
+
+  def set_for_test(self, hs_api):
+    self.history_server_api = hs_api
+
+  def app(self, appid):
+    common = {}
+
+    if self._executors and self._executors[0]:
+      common = self._massage_executor(self._executors[0])
+      common['properties'] = {
+          'metadata': [],
+          'counters': []
+      }
+      common['properties'].update(self._massage_executor(self._executors[0]))
+
+    return common
+
+  def _massage_executor(self, executor):
+    return {
+       "app_id": self.app_id,
+       "type": 'SPARK_EXECUTOR',
+       "id": self.app_executor_id,
+       "executor_id": executor['executor_id'],
+       "address": executor['address'],
+       "rdd_blocks": executor['rdd_blocks'],
+       "storage_memory": executor['storage_memory'],
+       "disk_used": executor['disk_used'],
+       "active_tasks": executor['active_tasks'],
+       "failed_tasks": executor['failed_tasks'],
+       "complete_tasks": executor['complete_tasks'],
+       "task_time": executor['task_time'],
+       "input": executor['input'],
+       "shuffle_read": executor['shuffle_read'],
+       "shuffle_write": executor['shuffle_write'],
+       "logs": executor['logs']
+    }
+
+  def logs(self, appid, app_type, log_name, offset=LOG_OFFSET_BYTES):
+    log = ""
+
+    if self._executors and self._executors[0]:
+      log = self.history_server_api.download_executor_logs(self.user, self._executors[0], log_name, offset)
+    return {
+       "logs": log
+    }

+ 129 - 4
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -410,6 +410,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
   <!-- ko if: type() == 'SPARK' -->
     <div data-bind="template: { name: 'job-spark-page${ SUFFIX }', data: $root.job() }"></div>
   <!-- /ko -->
+
+  <!-- ko if: type() == 'SPARK_EXECUTOR' -->
+    <div data-bind="template: { name: 'job-spark-executor-page${ SUFFIX }', data: $root.job() }"></div>
+  <!-- /ko -->
 </script>
 
 
@@ -802,7 +806,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
     <div data-bind="css:{'span10': !$root.isMini(), 'span12 no-margin': $root.isMini() }">
       <ul class="nav nav-pills margin-top-20">
-        <li class="active"><a href="#job-spark-page-properties${ SUFFIX }" data-toggle="tab">${ _('Properties') }</a></li>
+        <li class="active"><a class="job-spark-logs-link" href="#job-spark-page-logs${ SUFFIX }" data-toggle="tab">${ _('Logs') }</a></li>
+        <li><a href="#job-spark-page-executors${ SUFFIX }" data-bind="click: function(){ fetchProfile('executors'); $('a[href=\'#job-spark-page-executors${ SUFFIX }\']').tab('show'); }">${ _('Executors') }</a></li>
+        <li><a href="#job-spark-page-properties${ SUFFIX }" data-toggle="tab">${ _('Properties') }</a></li>
 
         <li class="pull-right" data-bind="template: { name: 'job-actions${ SUFFIX }' }"></li>
       </ul>
@@ -810,7 +816,56 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       <div class="clearfix"></div>
 
       <div class="tab-content">
-        <div class="tab-pane active" id="job-spark-page-properties${ SUFFIX }">
+        <div class="tab-pane active" id="job-spark-page-logs${ SUFFIX }">
+          <ul class="nav nav-tabs">
+          % 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-spark-page-executors${ SUFFIX }">
+          <form class="form-inline">
+            <input data-bind="textFilter: textFilter, clearable: {value: textFilter}, valueUpdate: 'afterkeydown'" type="text" class="input-xlarge search-query" placeholder="${_('Filter by name')}">
+          </form>
+
+          <table class="table table-condensed">
+            <thead>
+            <tr>
+              <th>${_('Executor Id')}</th>
+              <th>${_('Address')}</th>
+              <th>${_('RDD Blocks')}</th>
+              <th>${_('Storage Memory')}</th>
+              <th>${_('Disk Used')}</th>
+              <th>${_('Active Tasks')}</th>
+              <th>${_('Failed Tasks')}</th>
+              <th>${_('Complete Tasks')}</th>
+              <th>${_('Task Time')}</th>
+              <th>${_('Input')}</th>
+              <th>${_('Shuffle Read')}</th>
+              <th>${_('Shuffle Write')}</th>
+            </tr>
+            </thead>
+            <tbody data-bind="foreach: properties['executors']()['executor_list']">
+              <tr data-bind="click: function() { $root.job().id(id); $root.job().fetchJob(); }" class="status-border pointer">
+                <td data-bind="text: executor_id"></td>
+                <td data-bind="text: address"></td>
+                <td data-bind="text: rdd_blocks"></td>
+                <td data-bind="text: storage_memory"></td>
+                <td data-bind="text: disk_used"></td>
+                <td data-bind="text: active_tasks"></td>
+                <td data-bind="text: failed_tasks"></td>
+                <td data-bind="text: complete_tasks"></td>
+                <td data-bind="text: task_time"></td>
+                <td data-bind="text: input"></td>
+                <td data-bind="text: shuffle_read"></td>
+                <td data-bind="text: shuffle_write"></td>
+              </tr>
+            </tbody>
+          </table>
+        </div>
+        <div class="tab-pane" id="job-spark-page-properties${ SUFFIX }">
           <table class="datatables table table-condensed">
             <thead>
             <tr>
@@ -833,6 +888,69 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 </script>
 
 
+<script type="text/html" id="job-spark-executor-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">
+          <!-- ko with: properties -->
+          <li class="nav-header">${ _('Id') }</li>
+          <li class="break-word"><span data-bind="text: executor_id"></span></li>
+          <!-- /ko -->
+          <li class="nav-header">${ _('Type') }</li>
+          <li><span data-bind="text: type"></span></li>
+          <!-- ko if: !$root.isMini() -->
+          <!-- ko with: properties -->
+            <li class="nav-header">${ _('Address') }</li>
+            <li><span data-bind="text: address"></span></li>
+            <li class="nav-header">${ _('RDD Blocks') }</li>
+            <li><span data-bind="text: rdd_blocks"></span></li>
+            <li class="nav-header">${ _('Storage Memory') }</li>
+            <li><span data-bind="text: storage_memory"></span></li>
+            <li class="nav-header">${ _('Disk Used') }</li>
+            <li><span data-bind="text: disk_used"></span></li>
+            <li class="nav-header">${ _('Active Tasks') }</li>
+            <li><span data-bind="text: active_tasks"></span></li>
+            <li class="nav-header">${ _('Failed Tasks') }</li>
+            <li><span data-bind="text: failed_tasks"></span></li>
+            <li class="nav-header">${ _('Complet Tasks') }</li>
+            <li><span data-bind="text: complete_tasks"></span></li>
+            <li class="nav-header">${ _('Input') }</li>
+            <li><span data-bind="text: input"></span></li>
+            <li class="nav-header">${ _('Shuffle Read') }</li>
+            <li><span data-bind="text: shuffle_read"></span></li>
+            <li class="nav-header">${ _('Shuffle Write') }</li>
+            <li><span data-bind="text: shuffle_write"></span></li>
+          <!-- /ko -->
+          <!-- /ko -->
+        </ul>
+      </div>
+    </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-spark-executor-page-logs${ SUFFIX }" data-toggle="tab">${ _('Logs') }</a></li>
+      </ul>
+
+      <div class="tab-content">
+        <div class="tab-pane active" id="job-spark-executor-page-logs${ SUFFIX }">
+          <ul class="nav nav-tabs">
+          % 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>
+
+    </div>
+  </div>
+
+</script>
+
+
 <script type="text/html" id="dataeng-job-page${ SUFFIX }">
   <button class="btn" title="${ _('Troubleshoot') }" data-bind="click: troubleshoot">
     <i class="fa fa-tachometer"></i> ${ _('Troubleshoot') }
@@ -1717,7 +1835,6 @@ ${ 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());
 
@@ -1897,6 +2014,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             if (/^task_/.test(vm.job().id())) {
               crumbs.push({'id': vm.job().properties['app_id'], 'name': vm.job().properties['app_id'], 'type': 'app'});
             }
+            if (/_executor_/.test(vm.job().id())) {
+              crumbs.push({'id': vm.job().properties['app_id'], 'name': vm.job().properties['app_id'], 'type': 'app'});
+            }
             var oozieWorkflow = vm.job().name().match(/oozie:launcher:T=.+?:W=.+?:A=.+?:ID=(.+?-oozie-oozi-W)$/i);
             if (oozieWorkflow) {
               crumbs.push({'id': oozieWorkflow[1], 'name': oozieWorkflow[1], 'type': 'workflow'});
@@ -1919,7 +2039,12 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
               }
             }
 
-            crumbs.push({'id': vm.job().id(), 'name': vm.job().name(), 'type': vm.job().type()});
+            if (vm.job().type() == 'SPARK_EXECUTOR') {
+               crumbs.push({'id': vm.job().id(), 'name': vm.job().properties['executor_id'](), 'type': vm.job().type()});
+            }
+            else {
+               crumbs.push({'id': vm.job().id(), 'name': vm.job().name(), 'type': vm.job().type()});
+            }
             vm.resetBreadcrumbs(crumbs);
             //show is still bound to old job, setTimeout allows knockout model change event done at begining of this method to sends it's notification
             setTimeout(function () {

+ 166 - 0
apps/jobbrowser/src/jobbrowser/tests.py

@@ -34,14 +34,17 @@ from hadoop import cluster
 from hadoop.conf import YARN_CLUSTERS
 from hadoop.pseudo_hdfs4 import is_live_cluster
 from hadoop.yarn import resource_manager_api, mapreduce_api, history_server_api
+from hadoop.yarn.spark_history_server_api import SparkHistoryServerApi
 from liboozie.oozie_api_tests import OozieServerProvider
 from oozie.models import Workflow
 
 from jobbrowser import views
 from jobbrowser.api import get_api
 from jobbrowser.apis.query_api import QueryApi
+from jobbrowser.apis import job_api
 from jobbrowser.conf import SHARE_JOBS
 from jobbrowser.models import can_view_job, can_modify_job, LinkJobLogs
+from jobbrowser.yarn_models import SparkJob
 
 
 LOG = logging.getLogger(__name__)
@@ -614,6 +617,75 @@ class TestImpalaApi(object):
      assert_equal(response.get(key), value)
 
 
+class TestSparkNoHadoop(object):
+  def setUp(self):
+    self.c = make_logged_in_client(is_superuser=False)
+    grant_access("test", "test", "jobbrowser")
+    self.user = User.objects.get(username='test')
+
+    if not hasattr(job_api, 'old_NativeYarnApi'):
+      job_api.old_NativeYarnApi = job_api.YarnApi
+
+    if not hasattr(views, 'old_get_api'):
+      views.old_get_api = views.get_api
+
+    job_api.NativeYarnApi = MockYarnApi
+    views.get_api = MockYarnApi
+
+  def tearDown(self):
+    job_api.NativeYarnApi = getattr(job_api, 'old_NativeYarnApi')
+    views.get_api = getattr(views, 'old_get_api')
+
+  def test_spark_executor_logs(self):
+    # Spark job status is succeed
+    query_executor_data = {u'interface': [u'"jobs"'], u'app_id': [u'"driver_executor_application_1513618343677_0018"']}
+    resp_executor = self.c.post('/jobbrowser/api/job/jobs', query_executor_data)
+    response_executor = json.loads(resp_executor.content)
+    assert_equal(response_executor['status'], 0)
+    assert_equal(response_executor['app']['executor_id'], 'driver')
+
+    query_log_data = {u'interface': [u'"jobs"'], u'type': [u'"SPARK"'], u'app_id': [u'"application_1513618343677_0018"'], u'name': [u'"default"']}
+    resp_log = self.c.post('/jobbrowser/api/job/logs', query_log_data)
+    response_log = json.loads(resp_log.content)
+    assert_equal(response_log['status'], 0)
+    assert_equal(response_log['logs']['logs'], 'dummy_logs')
+
+    # Spark job status is running
+    query_executor_data = {u'interface': [u'"jobs"'], u'app_id': [u'"driver_executor_application_1513618343677_0020"']}
+    resp_executor = self.c.post('/jobbrowser/api/job/jobs', query_executor_data)
+    response_executor = json.loads(resp_executor.content)
+    assert_equal(response_executor['status'], 0)
+    assert_equal(response_executor['app']['executor_id'], 'driver')
+
+    query_log_data = {u'interface': [u'"jobs"'], u'type': [u'"SPARK"'], u'app_id': [u'"application_1513618343677_0020"'], u'name': [u'"default"']}
+    resp_log = self.c.post('/jobbrowser/api/job/logs', query_log_data)
+    response_log = json.loads(resp_log.content)
+    assert_equal(response_log['status'], 0)
+    assert_equal(response_log['logs']['logs'], 'dummy_logs')
+
+
+class MockYarnApi:
+  def __init__(self, user, jt=None):
+    self.user = user
+
+  def get_job(self, jobid):
+    return MockSparkJob(app_id=jobid)
+
+
+class MockSparkJob(SparkJob):
+  def __init__(self, app_id):
+    self.history_server_api = MockSparkHistoryApi()
+    self.jobId = app_id
+    self.trackingUrl = 'http://localhost:8088/proxy/' + app_id
+
+    if app_id == 'application_1513618343677_0018':
+      self.status = 'SUCCEEDED'
+    elif app_id == 'application_1513618343677_0020':
+      self.status = 'RUNNING'
+
+    self._get_metrics()
+
+
 class MockResourceManagerHaApi(object):
   """
   Mock the RM API.
@@ -974,6 +1046,100 @@ class MockMapreduceApi(MockMapreduce2Api):
       job['job']['id'] = job_id
       return job
 
+class MockSparkHistoryApi(SparkHistoryServerApi):
+  def __init__(self):
+    self.APPS = [{
+        "id": "application_1513618343677_0018",
+        "name": "Sleep15minPySpark",
+        "attempts": [ {
+          "attemptId": "1",
+          "startTime": "2017-12-20T20:25:19.672GMT",
+          "endTime": "2017-12-20T20:40:43.768GMT",
+          "sparkUser": "test",
+          "completed": True
+      }]
+    }, {
+        "id": "application_1513618343677_0020",
+        "name": "Sleep15minPySpark",
+        "attempts": [ {
+          "attemptId": "2",
+          "startTime": "2017-12-24T03:19:29.993GMT",
+          "endTime": "1969-12-31T23:59:59.999GMT",
+          "sparkUser": "test",
+          "completed": False
+        }, {
+          "attemptId": "1",
+          "startTime": "2017-12-24T03:12:50.763GMT",
+          "endTime": "2017-12-24T03:19:22.178GMT",
+          "sparkUser": "test",
+          "completed": True
+      }]
+    }]
+
+  def applications(self):
+    return self.APPS
+
+  def executors(self, job):
+    EXECUTORS_LISTS = {
+      u'application_1513618343677_0018': [{
+        u'diskUsed': 0,
+        u'totalShuffleWrite': 0,
+        u'totalCores': 0,
+        u'executorLogs': {
+          u'stderr': u'http://localhost:8042/node/containerlogs/container_1513618343677_0018_01_000001/test/stderr?start=-4096',
+          u'stdout': u'http://localhost:8042/node/containerlogs/container_1513618343677_0018_01_000001/test/stdout?start=-4096'
+        },
+        u'totalInputBytes': 0,
+        u'rddBlocks': 0,
+        u'maxMemory': 515553361,
+        u'totalShuffleRead': 0,
+        u'totalTasks': 0,
+        u'activeTasks': 0,
+        u'failedTasks': 0,
+        u'completedTasks': 0,
+        u'hostPort': u'172.31.122.54:43234',
+        u'maxTasks': 0, u'totalGCTime': 0,
+        u'isBlacklisted': False,
+        u'memoryUsed': 0,
+        u'id': u'driver',
+        u'isActive': True,
+        u'totalDuration': 0
+      }],
+      u'application_1513618343677_0020/2' : [{
+        u'diskUsed': 0,
+        u'totalShuffleWrite': 0,
+        u'totalCores': 0,
+        u'executorLogs': {
+          u'stderr': u'http://localhost:8042/node/containerlogs/container_1513618343677_0020_01_000001/test/stderr?start=-4096',
+          u'stdout': u'http://localhost:8042/node/containerlogs/container_1513618343677_0020_01_000001/test/stdout?start=-4096'},
+        u'totalInputBytes': 0,
+        u'rddBlocks': 0,
+        u'maxMemory': 515553361,
+        u'totalShuffleRead': 0,
+        u'totalTasks': 0,
+        u'activeTasks': 0,
+        u'failedTasks': 0,
+        u'completedTasks': 0,
+        u'hostPort': u'172.31.122.65:38210',
+        u'maxTasks': 0,
+        u'totalGCTime': 0,
+        u'isBlacklisted': False,
+        u'memoryUsed': 0,
+        u'id': u'driver',
+        u'isActive': True,
+        u'totalDuration': 0}]
+    }
+    app_id = self.get_real_app_id(job)
+    if not app_id:
+      return []
+
+    return EXECUTORS_LISTS[app_id] if app_id in EXECUTORS_LISTS else []
+
+  def download_executors_logs(self, request, job, name, offset):
+    return 'dummy_logs'
+
+  def download_executor_logs(self, user, executor, name, offset):
+    return 'dummy_log'
 
 class HistoryServerApi(MockMapreduce2Api):
 

+ 17 - 0
apps/jobbrowser/src/jobbrowser/views.py

@@ -301,6 +301,20 @@ def kill_job(request, job):
 
   raise Exception(_("Job did not appear as killed within 15 seconds."))
 
+@check_job_permission
+def job_executor_logs(request, job, attempt_index=0, name='syslog', offset=LOG_OFFSET_BYTES):
+  response = {'status': -1}
+  try:
+    log = ''
+    if job.status not in ('NEW', 'SUBMITTED', 'ACCEPTED'):
+      log = job.history_server_api.download_executors_logs(request, job, name, offset)
+    response['status'] = 0
+    response['log'] = LinkJobLogs._make_hdfs_links(log)
+  except Exception, e:
+    response['log'] = _('Failed to retrieve executor log: %s' % e)
+
+  return JsonResponse(response)
+
 
 @check_job_permission
 def job_attempt_logs(request, job, attempt_index=0):
@@ -378,6 +392,9 @@ def job_single_logs(request, job, offset=LOG_OFFSET_BYTES):
   def cmp_exec_time(task1, task2):
     return cmp(task1.execStartTimeMs, task2.execStartTimeMs)
 
+  if job.applicationType == 'SPARK':
+    return job.history_server_api.download_logs(job.app)
+
   task = None
 
   failed_tasks = job.filter_tasks(task_states=('failed',))

+ 18 - 3
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -31,6 +31,8 @@ from desktop.lib.view_util import big_filesizeformat, format_duration_in_millis
 
 from hadoop.yarn.clients import get_log_client
 
+from itertools import izip
+
 from jobbrowser.models import format_unixtime_ms
 
 
@@ -111,13 +113,14 @@ class SparkJob(Application):
   def __init__(self, job, rm_api=None, hs_api=None):
     super(SparkJob, self).__init__(job, rm_api)
     self._resolve_tracking_url()
-    if self.state not in ('NEW', 'SUBMITTED', 'ACCEPTED', 'RUNNING') and hs_api:
+    if self.status not in ('NEW', 'SUBMITTED', 'ACCEPTED') and hs_api:
       self.history_server_api = hs_api
       self._get_metrics()
 
   @property
   def logs_url(self):
-    return os.path.join(self.trackingUrl, 'executors')
+    log_links = self.history_server_api.get_executors_loglinks(self)
+    return log_links['stdout'] if log_links and 'stdout' in log_links else ''
 
   @property
   def attempt_id(self):
@@ -140,7 +143,7 @@ class SparkJob(Application):
   def _get_metrics(self):
     self.metrics = {}
     try:
-      executors = self.history_server_api.executors(self.jobId, self.attempt_id)
+      executors = self.history_server_api.executors(self)
       if executors:
         self.metrics['headers'] = [
           _('Executor Id'),
@@ -177,6 +180,18 @@ class SparkJob(Application):
       LOG.error('Failed to get Spark Job executors: %s' % e)
       # Prevent a nosedive. Don't create metrics if api changes or url is unreachable.
 
+  def get_executors(self):
+    executor_list = []
+    if hasattr(self, 'metrics') and 'executors' in self.metrics:
+      executors = self.metrics['executors']
+      headers = ['executor_id', 'address', 'rdd_blocks', 'storage_memory', 'disk_used', 'active_tasks', 'failed_tasks',
+                 'complete_tasks', 'task_time', 'input', 'shuffle_read', 'shuffle_write', 'logs']
+      for executor in executors:
+        executor_data = dict(izip(headers, executor))
+        executor_data.update({'id': executor_data['executor_id'] + '_executor_' + self.jobId, 'type': 'SPARK_EXECUTOR'})
+        executor_list.append(executor_data)
+    return executor_list
+
 
 class Job(object):
 

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

@@ -124,6 +124,10 @@ class ResourceManagerApi(object):
     params = self._get_params()
     return self._execute(self._root.get, 'cluster/apps/%(app_id)s' % {'app_id': app_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
 
+  def appattempts(self, app_id):
+    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 kill(self, app_id):
     data = {'state': 'KILLED'}
     token = None

+ 77 - 1
desktop/libs/hadoop/src/hadoop/yarn/spark_history_server_api.py

@@ -15,14 +15,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import logging
 import posixpath
 import threading
+import urlparse
 
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.rest.http_client import HttpClient
 from desktop.lib.rest.resource import Resource
+from django.utils.translation import ugettext as _
 from hadoop import cluster
+from hadoop.yarn.clients import get_log_client
+
+from lxml import html
 
 
 LOG = logging.getLogger(__name__)
@@ -93,7 +99,11 @@ class SparkHistoryServerApi(object):
   def stages(self, app_id):
     return self._root.get('applications/%(app_id)s/stages' % {'app_id': app_id}, headers=self.headers)
 
-  def executors(self, app_id):
+  def executors(self, job):
+    app_id = self.get_real_app_id(job)
+    if not app_id:
+      return []
+
     return self._root.get('applications/%(app_id)s/executors' % {'app_id': app_id}, headers=self.headers)
 
   def stage_attempts(self, app_id, stage_id):
@@ -119,3 +129,69 @@ class SparkHistoryServerApi(object):
 
   def download_attempt_logs(self, app_id, attempt_id):
     return self._root.get('applications/%(app_id)s/%(attempt_id)s/logs' % {'app_id': app_id, 'attempt_id': attempt_id}, headers=self.headers)
+
+  def download_executors_logs(self, request, job, name, offset):
+    log_links = self.get_executors_loglinks(job)
+
+    return self.retrieve_log_content(log_links, name, request.user.username, offset)
+
+  def download_executor_logs(self, user, executor, name, offset):
+    return self.retrieve_log_content(executor['logs'], name, user.username, offset)
+
+  def retrieve_log_content(self, log_links, log_name, username, offset):
+    params = {
+      'doAs': username
+    }
+
+    if offset != 0:
+      params['start'] = offset
+
+    if not log_name or not log_name == 'stderr':
+      log_name = 'stdout'
+
+    log = ''
+    if log_links and log_name in log_links:
+      log_link = log_links[log_name]
+
+      root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
+      response = root.get('', params=params)
+      log = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
+    return log
+
+  def get_executors_loglinks(self, job):
+    if job.metrics and 'executors' in job.metrics and job.metrics['executors']:
+      executors = [executor for executor in job.metrics['executors'] if executor[0] == 'driver']  # look up driver executor
+      if not executors:
+        executor = job.metrics['executors'][0]
+      else:
+        executor = executors[0]
+
+    return None if not executor else executor[12]
+
+  def get_real_app_id(self, job):
+    # https://spark.apache.org/docs/1.6.0/monitoring.html and https://spark.apache.org/docs/2.0.0/monitoring.html
+    # When running on Yarn, each application has multiple attempts, so [app-id] is actually [app-id]/[attempt-id] in all cases.
+    # When running job as cluster mode, an attempt number is part of application ID, but proxy URL can't be resolved to match
+    # Spark history URL. In the applications list, each job's attampt list shows if attempt ID is used and how many attempts.
+
+    app_id = job.jobId if job.jobId == job.attempt_id else job.jobId + '/' + job.attempt_id
+    if job.status not in ('SUCCEEDED', 'FAILED'):
+      try:
+        jobs_json = self.applications()
+        job_filtered_json = [x for x in jobs_json if x['id'] == job.jobId]
+
+        if not job_filtered_json:
+          return {}
+
+        attempts = job_filtered_json[0]['attempts']
+
+        if len(attempts) == 1:
+          app_id = job.jobId if 'attemptId' not in attempts[0] else job.jobId + '/' + attempts[0]['attemptId']
+        else:
+          app_id = job.jobId + '/%d' % len(attempts)
+
+      except Exception as e:
+        LOG.error('Cannot get executors %s' % e)
+        app_id = None
+
+    return app_id