Browse Source

HUE-1176 [jb] Open task of a MapReduce job

Romain Rigaux 9 năm trước cách đây
mục cha
commit
f17f0483d6

+ 1 - 1
apps/jobbrowser/src/jobbrowser/api.py

@@ -304,7 +304,7 @@ class YarnApi(JobBrowserApi):
     return self.get_job(jobid).filter_tasks(**filters)
 
   def get_task(self, jobid, task_id):
-    return self.get_job(jobid).task(task_id)
+    return self.get_job(jobid).get_task(task_id)
 
   def get_tracker(self, node_manager_http_address, container_id):
     api = node_manager_api.get_node_manager_api('http://' + node_manager_http_address)

+ 2 - 2
apps/jobbrowser/src/jobbrowser/apis/base_api.py

@@ -27,11 +27,11 @@ LOG = logging.getLogger(__name__)
 
 def get_api(user, interface):
   from jobbrowser.apis.workflow_api import WorkflowApi
-  from jobbrowser.apis.job_api import YarnApi
+  from jobbrowser.apis.job_api import JobApi
   from jobbrowser.apis.schedule_api import ScheduleApi
 
   if interface == 'apps':
-    return YarnApi(user)
+    return JobApi(user)
   elif interface == 'workflows':
     return WorkflowApi(user)
   elif interface == 'schedules':

+ 61 - 9
apps/jobbrowser/src/jobbrowser/apis/job_api.py

@@ -36,13 +36,13 @@ except Exception, e:
 class JobApi(Api):
 
   def __init__(self, user):
-    self.user =  user
+    self.user = user
     self.yarn_api = YarnApi(user) # TODO: actually long term move job aggregations to the frontend instead probably
     self.impala_api = ImpalaApi(user)
     self.request = None
 
   def apps(self):
-    jobs = self.self.yarn_api.apps()
+    jobs = self.yarn_api.apps()
     # += Impala
     # += Sqoop2
     return jobs
@@ -50,14 +50,20 @@ class JobApi(Api):
   def app(self, appid):
     return self._get_api(appid).app(appid)
 
-  def logs(self, appid):
-    return self._get_api(appid).logs(appid)
+  def logs(self, appid, app_type):
+    return self._get_api(appid).logs(appid, app_type)
 
   def profile(self, appid, app_type, app_property):
     return self._get_api(appid).profile(appid, app_type, app_property)
 
   def _get_api(self, appid):
-    return self.impala_api if not appid.startswith('application_') else self.yarn_api
+    if appid.startswith('task_'):
+      return YarnMapReduceTaskApi(self.user, appid)
+    else:
+      return self.yarn_api # application_
+
+  def _set_request(self, request):
+    self.request = request
 
 
 class YarnApi(Api):
@@ -120,14 +126,60 @@ class YarnApi(Api):
   def profile(self, appid, app_type, app_property):
     if app_type == 'MAPREDUCE':
       if app_property == 'tasks':
-        response = tasks(self.request, job=appid)
-        return json.loads(response.content)
+        return {
+          'task_list': YarnMapReduceTaskApi(self.user, appid).apps(),
+        }
 
     return {}
 
 
-  def _set_request(self, request):
-    self.request = request
+class YarnMapReduceTaskApi(Api):
+
+  def __init__(self, user, app_id):
+    Api.__init__(self, user)
+    self.app_id = '_'.join(app_id.replace('task_', 'application_').split('_')[:3])
+
+
+  def apps(self):
+    return [self._massage_task(task) for task in NativeYarnApi(self.user).get_tasks(jobid=self.app_id, pagenum=1)]
+
+  def app(self, appid):
+    task = NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=appid)
+
+    common = self._massage_task(task)
+    common['properties'] = {
+    }
+
+    return common
+
+
+  def logs(self, appid, app_type):
+    if app_type == 'MAPREDUCE':
+      response = job_attempt_logs_json(MockDjangoRequest(self.user), job=appid)
+      logs = json.loads(response.content)['log']
+    else:
+      logs = None
+    return {'progress': 0, 'logs': {'default': logs}}
+
+
+  def profile(self, appid, app_type, app_property):
+    if app_property == 'task_attemps':
+      response = tasks(self.request, job=appid)
+      return json.loads(response.content)
+
+    return {}
+
+  def _massage_task(self, task):
+    return {
+        'id': task.id,
+        'type': task.type,
+        'elapsedTime': task.elapsedTime,
+        'progress': task.progress,
+        'state': task.state,
+        'startTime': task.startTime,
+        'successfulAttempt': task.successfulAttempt,
+        'finishTime': task.finishTime
+    }
 
 
 class YarnAtsApi(Api):

+ 39 - 8
apps/jobbrowser/src/jobbrowser/templates/apps.mako

@@ -116,13 +116,14 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         </div>
         <div class="resizer" data-bind="visible: $root.isLeftPanelVisible() && $root.assistAvailable(), splitDraggable : { appName: 'notebook', leftPanelVisible: $root.isLeftPanelVisible }"><div class="resize-bar">&nbsp;</div></div>
         % endif
+
         <div class="content-panel">
 
           <div class="container-fluid">
             <div data-bind="template: { name: 'breadcrumbs' }"></div>
 
             <!-- ko if: ! $root.job() -->
-            ${_('Filter')} <input id="textFilter" type="text" class="input-xlarge search-query" placeholder="${_('Filter by id, name, user...')}" value="user:${ user.username }">
+            ${_('Filter')} <input type="text" class="input-xlarge search-query" placeholder="${_('Filter by id, name, user...')}" value="user:${ user.username }">
             <span class="btn-group">
               <class="btn-group">
                 <a class="btn btn-status btn-success" data-value="completed">${ _('Succeeded') }</a>
@@ -231,6 +232,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
     <div data-bind="template: { name: 'job-mapreduce-page', data: $root.job() }"></div>
   <!-- /ko -->
 
+  <!-- ko if: type() == 'MAP' || type() == 'REDUCE' -->
+    <div data-bind="template: { name: 'job-mapreduce-task-page', data: $root.job() }"></div>
+  <!-- /ko -->
+
   <!-- ko if: type() == 'YARN' -->
     <div data-bind="template: { name: 'job-yarn-page', data: $root.job() }"></div>
   <!-- /ko -->
@@ -305,6 +310,20 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
     <div class="tab-pane" id="job-mapreduce-page-tasks">
       <a href="javascript:void(0)" data-bind="click: function() { return fetchProfile('tasks'); }">Load</a>
+      </br>
+
+      ${_('Filter')} <input type="text" class="input-xlarge search-query" placeholder="${_('Filter by id, name, user...')}" value="user:${ user.username }">
+      <span class="btn-group">
+        <class="btn-group">
+          <a class="btn btn-status btn-success" data-value="completed">${ _('MAP') }</a>
+          <a class="btn btn-status btn-warning" data-value="running">${ _('REDUCE') }</a>
+        </span>
+      </span>
+
+      <div class="btn-toolbar pull-right" style="display: inline; vertical-align: middle; margin-left: 10px; font-size: 12px">
+        <span class="loader hide"><i class="fa fa-2x fa-spinner fa-spin muted"></i></span>
+        <button class="btn bulkToolbarBtn bulk-resume" data-operation="resume" title="${ _('Resume selected') }" disabled="disabled" type="button"><i class="fa fa-play"></i><span class="hide-small"> ${ _('View') }</span></button>
+      </div>
 
       <table class="table table-condensed">
         <thead>
@@ -312,17 +331,25 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           <th width="1%"><div class="select-all hueCheckbox fa"></div></th>
           <th>${_('Type')}</th>
           <th>${_('Id')}</th>
-          <th>${_('Logs')}</th>
-          <th>${_('Url')}</th>
+          <th>${_('elapsedTime')}</th>
+          <th>${_('progress')}</th>
+          <th>${_('state')}</th>
+          <th>${_('startTime')}</th>
+          <th>${_('successfulAttempt')}</th>
+          <th>${_('finishTime')}</th>
         </tr>
         </thead>
         <tbody data-bind="foreach: properties['tasks']()['task_list']">
-          <tr data-bind="click: function() { $root.job().fetchJob(id); }">
+          <tr data-bind="click: function() { $root.job().id(id); $root.job().fetchJob(); }">
             <td><div class="hueCheckbox fa"></div></td>
             <td data-bind="text: type"></td>
             <td data-bind="text: id"></td>
-            <td data-bind="text: logs"></td>
-            <td data-bind="text: url"></td>
+            <td data-bind="text: elapsedTime"></td>
+            <td data-bind="text: progress"></td>
+            <td data-bind="text: state"></td>
+            <td data-bind="text: startTime"></td>
+            <td data-bind="text: successfulAttempt"></td>
+            <td data-bind="text: finishTime"></td>
           </tr>
         </tbody>
       </table>
@@ -345,14 +372,18 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
   <h2>MapReduce Task</h2>
   <br/>
 
-  Task Attemps | Metadata | Counters
+  Attempts | Metadata | Counters
 </script>
 
 <script type="text/html" id="job-mapreduce-task-attempt-page">
   <h2>MapReduce Task attempt</h2>
   <br/>
 
-  Container | Metadata | Counters
+  Logs | Metadata | Counters
+
+  <br/><br/>
+  Attempt | Container
+
 </script>
 
 

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

@@ -425,15 +425,6 @@ def tasks(request, job):
 
   filter_params = copy_query_dict(request.GET, ('tasktype', 'taskstate', 'tasktext')).urlencode()
 
-  if request.is_ajax():
-    return JsonResponse({
-      'filter_params': filter_params,
-      'task_list': [massage_task_for_json(task) for task in task_list],
-      'tasktype': ttypes,
-      'taskstate': tstates,
-      'tasktext': ttext
-    })
-
   return render("tasks.mako", request, {
     'request': request,
     'filter_params': filter_params,