Browse Source

HUE-1157 [jb] MR2 Container link broken

Romain Rigaux 12 years ago
parent
commit
557cfc1

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

@@ -25,10 +25,11 @@ from hadoop.api.jobtracker.ttypes import ThriftJobPriority, TaskTrackerNotFoundE
 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
+import hadoop.yarn.node_manager_api as node_manager_api
 
 from jobbrowser.conf import SHARE_JOBS
-from jobbrowser.models import Job, JobLinkage, TaskList
-from jobbrowser.yarn_models import Application, Job as YarnJob
+from jobbrowser.models import Job, JobLinkage, TaskList, Tracker
+from jobbrowser.yarn_models import Application, Job as YarnJob, Container
 
 
 LOG = logging.getLogger(__name__)
@@ -140,6 +141,9 @@ class JtApi(JobBrowserApi):
                            _DEFAULT_OBJ_PER_PAGINATION,
                            _DEFAULT_OBJ_PER_PAGINATION * (filters['pagenum'] - 1))
 
+  def get_tracker(self, trackerid):
+    return Tracker.from_name(self.jt, trackerid)
+
 
 class YarnApi(JobBrowserApi):
   """
@@ -158,6 +162,7 @@ class YarnApi(JobBrowserApi):
     self.user = user
     self.resource_manager_api = resource_manager_api.get_resource_manager()
     self.mapreduce_api = mapreduce_api.get_mapreduce_api()
+    self.node_manager_api = node_manager_api.get_resource_manager_api()
     self.history_server_api = history_server_api.get_history_server_api()
 
   def get_job_link(self, job_id):
@@ -216,3 +221,6 @@ class YarnApi(JobBrowserApi):
 
   def get_task(self, jobid, task_id):
     return self.get_job(jobid).task(task_id)
+
+  def get_tracker(self, container_id):
+    return Container(self.node_manager_api.container(container_id))

+ 0 - 0
apps/jobbrowser/src/jobbrowser/lib/__init__.py


+ 7 - 1
apps/jobbrowser/src/jobbrowser/templates/attempt.mako

@@ -101,7 +101,13 @@ ${ commonheader(_('Task Attempt: %(attemptId)s - Job Browser') % dict(attemptId=
                         </tr>
                         <tr>
                             <td>${_('Task Tracker')}</td>
-                            <td><a href="/jobbrowser/trackers/${attempt.taskTrackerId}">${attempt.taskTrackerId}</a></td>
+                            <td>
+                              % if attempt.is_mr2:
+                                ${ comps.get_container_link(status, attempt.taskTrackerId) }
+                              % else:
+                                <a href="/jobbrowser/trackers/${attempt.taskTrackerId}" class="task_tracker_link">${attempt.taskTrackerId}</a>
+                              % endif
+                            </td>
                         </tr>
                         <tr>
                             <td>${_('Phase')}</td>

+ 1 - 1
apps/jobbrowser/src/jobbrowser/templates/job.mako

@@ -182,7 +182,7 @@ ${ commonheader(_('Job: %(jobId)s - Job Browser') % dict(jobId=job.jobId_short),
                                             </a>
                                         </td>
                                         <td>${ attempt['id'] }</td>
-                                        <td>${ attempt['containerId'] }</td>
+                                        <td>${ comps.get_container_link(job.status, attempt['containerId']) }</td>
                                     </tr>
                                 % endfor
                             </tbody>

+ 8 - 0
apps/jobbrowser/src/jobbrowser/templates/jobbrowser_components.mako

@@ -191,6 +191,14 @@
     % endif
 </%def>
 
+<%def name="get_container_link(status, container_id)">
+    ## As soon as the job finishes the container disappears
+    % if status.lower() in ('running', 'prep', 'accepted', 'finishing'):
+        <a href="${ url('jobbrowser.views.single_tracker',trackerid=container_id) }" class="task_tracker_link">${ container_id }</a>
+    % else:
+        ${ container_id }
+    % endif
+</%def>
 
 <%def name="get_bootstrap_class(job, prefix)">
     <%

+ 7 - 1
apps/jobbrowser/src/jobbrowser/templates/task.mako

@@ -78,7 +78,13 @@ ${ commonheader(_('Job Task: %(taskId)s - Job Browser') % dict(taskId=task.taskI
                                            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>
+                                    <td>
+                                      % if task.is_mr2:
+                                        ${ comps.get_container_link(task.state, attempt.taskTrackerId) }
+                                      % else:
+                                        <a href="/jobbrowser/trackers/${attempt.taskTrackerId}" class="task_tracker_link">${attempt.taskTrackerId}</a>
+                                      % endif
+                                    </td>
                                     <td>${attempt.startTimeFormatted}</td>
                                     <td>${attempt.finishTimeFormatted}</td>
                                     <td>${attempt.outputSize}</td>

+ 58 - 34
apps/jobbrowser/src/jobbrowser/templates/tasktracker.mako

@@ -17,45 +17,69 @@
   from desktop.views import commonheader, commonfooter
   from django.utils.translation import ugettext as _
 %>
+
 <%namespace name="comps" file="jobbrowser_components.mako" />
 
 ${ commonheader(_('Tracker: %(trackerId)s - Job Browser') % dict(trackerId=tracker.trackerId), "jobbrowser", user) | n,unicode }
+
 <div class="container-fluid">
-	<h1>${_('Tracker at %(trackerHost)s on port %(trackerPort)s') % dict(trackerHost=tracker.host, trackerPort=tracker.httpPort)}</h1>
-	<div>
-		<dl>
-			<dt>${_('ID')}</dt>
-			<dd>${ tracker.trackerId }</dd>
-			<dt>${_('Last heard from at')}</dt>
-			<dd>${ tracker.lastSeenFormatted }.</dd>
-		</dl>
-	</div>
-	
-	<h2>${_('Memory Metrics')}</h2>
-	<div>
-		<dl>
-			<dt>${_('Total virtual memory:')}</dt>
-			<dd>${tracker.totalVirtualMemory }</dd>
-			<dt>${_('Total physical memory:')}</dt>
-			<dd>${tracker.totalPhysicalMemory }</dd>
-			<dt>${_('Available space:')}</dt>
-			<dd>${tracker.availableSpace}</dd>
-		</dl>
-	</div>
+    <h1>${_('Tracker at %(trackerHost)s on port %(trackerPort)s') % dict(trackerHost=tracker.host, trackerPort=tracker.httpPort)}</h1>
+    <div>
+        <dl>
+            <dt>${_('ID')}</dt>
+            <dd>${ tracker.trackerId }</dd>
+            % if not tracker.is_mr2:
+            <dt>${_('Last heard from at')}</dt>
+            <dd>${ tracker.lastSeenFormatted }.</dd>
+            % endif
+        </dl>
+    </div>
+
+    % if tracker.is_mr2:
+    <h2>${_('Memory Metrics')}</h2>
+    <div>
+        <dl>
+            <dt>${_('Node Id')}</dt>
+            <dd>${tracker.nodeId }</dd>
+            <dt>${_('State')}</dt>
+            <dd>${tracker.state }</dd>
+            <dt>${_('User')}</dt>
+            <dd>${tracker.user}</dd>
+            <dt>${_('Diagnostics')}</dt>
+            <dd>${tracker.diagnostics}</dd>
+            <dt>${_('Total Memory Needed in MB')}</dt>
+            <dd>${tracker.totalMemoryNeededMB}</dd>
+            <dt>${_('Exit Code')}</dt>
+            <dd>${tracker.exitCode}</dd>
+        </dl>
+    </div>
+    % else:
+    <h2>${_('Memory Metrics')}</h2>
+    <div>
+        <dl>
+            <dt>${_('Total virtual memory:')}</dt>
+            <dd>${tracker.totalVirtualMemory }</dd>
+            <dt>${_('Total physical memory:')}</dt>
+            <dd>${tracker.totalPhysicalMemory }</dd>
+            <dt>${_('Available space:')}</dt>
+            <dd>${tracker.availableSpace}</dd>
+        </dl>
+    </div>
 
-	<h2>${_('Map and Reduce')}</h2>
-	<div>
-		<dl>
-			<dt>${_('Map count:')}</dt>
-			<dd>${tracker.mapCount}</dd>
-			<dt>${_('Reduce count:')}</dt>
-			<dd>${tracker.reduceCount}</dd>
-			<dt>${_('Max map tasks:')}</dt>
-			<dd>${tracker.maxMapTasks}</dd>
-			<dt>${_('Max reduce tasks:')}</dt>
-			<dd>${tracker.maxReduceTasks}</dd>
-		</dl>
-	</div>
+    <h2>${_('Map and Reduce')}</h2>
+    <div>
+        <dl>
+            <dt>${_('Map count:')}</dt>
+            <dd>${tracker.mapCount}</dd>
+            <dt>${_('Reduce count:')}</dt>
+            <dd>${tracker.reduceCount}</dd>
+            <dt>${_('Max map tasks:')}</dt>
+            <dd>${tracker.maxMapTasks}</dd>
+            <dt>${_('Max reduce tasks:')}</dt>
+            <dd>${tracker.maxReduceTasks}</dd>
+        </dl>
+    </div>
+    % endif
 </div>
 
 ${ commonfooter(messages) | n,unicode }

+ 3 - 3
apps/jobbrowser/src/jobbrowser/urls.py

@@ -27,18 +27,18 @@ urlpatterns = patterns('jobbrowser.views',
   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+)/job_attempt_logs_json/(?P<attempt_index>\d+)/(?P<name>\w+)?/(?P<offset>\d+)?$', 'job_attempt_logs_json', name='job_attempt_logs_json'), # MR2
   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'),
+  url(r'^trackers/(?P<trackerid>.+)$', 'single_tracker', name='single_tracker'),
   # Unused
+  url(r'^trackers$', 'trackers', name='trackers'),
   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'),
 )

+ 6 - 8
apps/jobbrowser/src/jobbrowser/views.py

@@ -258,8 +258,7 @@ def single_task_attempt(request, job, taskid, attemptid):
   except (KeyError, RestException), e:
     raise PopupException(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
 
-  return render("attempt.mako", request,
-    {
+  return render("attempt.mako", request, {
       "attempt": attempt,
       "taskid": taskid,
       "joblnk": job_link,
@@ -297,7 +296,7 @@ def single_task_attempt_logs(request, job, taskid, attemptid):
     # Four entries,
     # for diagnostic, stdout, stderr and syslog
     logs = [ _("Failed to retrieve log. TaskTracker not found.") ] * 4
-  
+
   context = {
       "attempt": attempt,
       "taskid": taskid,
@@ -306,7 +305,7 @@ def single_task_attempt_logs(request, job, taskid, attemptid):
       "logs": logs,
       "first_log_tab": first_log_tab,
   }
-  
+
   if request.GET.get('format') == 'python':
     return context
   else:
@@ -344,10 +343,9 @@ def trackers(request):
   return render("tasktrackers.mako", request, {'trackers':trackers})
 
 def single_tracker(request, trackerid):
-  """
-  We get here from /trackers/trackerid
-  """
-  tracker = Tracker.from_name(request.jt, trackerid)
+  jt = get_api(request.user, request.jt)
+
+  tracker = jt.get_tracker(trackerid)
   return render("tasktracker.mako", request, {'tracker':tracker})
 
 def clusterstatus(request):

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

@@ -204,3 +204,30 @@ class Attempt:
     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
+
+
+class Container:
+
+  def __init__(self, attrs):
+    if attrs:
+      for key, value in attrs['container'].iteritems():
+        setattr(self, key, value)
+    self.is_mr2 = True
+
+    self._fixup()
+
+  def _fixup(self):
+    setattr(self, 'trackerId', self.id)
+    setattr(self, 'httpPort', self.nodeId.split(':')[1])
+    setattr(self, 'host', self.nodeId.split(':')[0])
+    setattr(self, 'lastSeenMs', None)
+    setattr(self, 'lastSeenFormatted', '')
+    setattr(self, 'totalVirtualMemory', None)
+    setattr(self, 'totalPhysicalMemory', self.totalMemoryNeededMB)
+    setattr(self, 'availableSpace', None)
+    setattr(self, 'failureCount', None)
+    setattr(self, 'mapCount', None)
+    setattr(self, 'reduceCount', None)
+    setattr(self, 'maxMapTasks', None)
+    setattr(self, 'maxReduceTasks', None)
+    setattr(self, 'taskReports', None)

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

@@ -326,6 +326,9 @@
       # URL of the HistoryServer API
       history_server_api_url=http://localhost:19888
 
+      # URL of the NodeManager API
+      node_manager_api_url=http://localhost:8042
+
 
 ###########################################################################
 # Settings to configure liboozie

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

@@ -329,6 +329,9 @@
       # URL of the HistoryServer API
       history_server_api_url=http://localhost:19888
 
+      # URL of the NodeManager API
+      node_manager_api_url=http://localhost:8042
+
 
 ###########################################################################
 # Settings to configure liboozie

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

@@ -219,6 +219,9 @@ YARN_CLUSTERS = UnspecifiedConfigSection(
       HISTORY_SERVER_API_URL=Config("history_server_api_url",
                   default='http://localhost:19888',
                   help="URL of the HistoryServer API"),
+      NODE_MANAGER_API_URL=Config("node_manager_api_url",
+                  default='http://localhost:8042',
+                  help="URL of the HistoryServer API"),
     )
   )
 )

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

@@ -95,3 +95,6 @@ class MapreduceApi(object):
     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})
 
+  def task_attempt(self, job_id, task_id, attempt_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/%(attempt_id)s' % {'app_id': app_id, 'job_id': job_id, 'task_id': task_id, 'attempt_id': attempt_id, 'version': _API_VERSION}, headers={'Accept': _JSON_CONTENT_TYPE})

+ 72 - 0
desktop/libs/hadoop/src/hadoop/yarn/node_manager_api.py

@@ -0,0 +1,72 @@
+#!/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_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 = ResourceManagerApi(yarn_cluster.NODE_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 "NodeManagerApi at %s" % (self._url,)
+
+  @property
+  def url(self):
+    return self._url
+
+  @property
+  def security_enabled(self):
+    return self._security_enabled
+
+  def containers(self):
+    return self._root.get('node/containers', headers={'Accept': _JSON_CONTENT_TYPE})
+
+  def container(self, container_id):
+    return self._root.get('node/containers/%(container_id)s' % {'container_id': container_id}, headers={'Accept': _JSON_CONTENT_TYPE})