Эх сурвалжийг харах

HUE-3434 [jb] Logs of finished Oozie workflow are not displayed

Jenny Kim 9 жил өмнө
parent
commit
3a210c4ff8

+ 8 - 8
apps/jobbrowser/src/jobbrowser/models.py

@@ -38,7 +38,7 @@ from desktop.lib.exceptions_renderable import PopupException
 from django.utils.translation import ugettext as _
 from jobbrowser.conf import DISABLE_KILLING_JOBS
 
-LOGGER = logging.getLogger(__name__)
+LOG = logging.getLogger(__name__)
 
 
 def can_view_job(username, job):
@@ -449,11 +449,11 @@ class TaskAttempt(object):
       tracker = Tracker.from_name(self.task.jt, self.taskTrackerId)
       return tracker
     except ttypes.TaskTrackerNotFoundException, e:
-      LOGGER.warn("Tracker %s not found: %s" % (self.taskTrackerId, e))
-      if LOGGER.isEnabledFor(logging.DEBUG):
+      LOG.warn("Tracker %s not found: %s" % (self.taskTrackerId, e))
+      if LOG.isEnabledFor(logging.DEBUG):
         all_trackers = self.task.jt.all_task_trackers()
         for t in all_trackers.trackers:
-          LOGGER.debug("Available tracker: %s" % (t.trackerName,))
+          LOG.debug("Available tracker: %s" % (t.trackerName,))
       raise ttypes.TaskTrackerNotFoundException(
                           _("Cannot look up TaskTracker %(id)s.") % {'id': self.taskTrackerId})
 
@@ -475,7 +475,7 @@ class TaskAttempt(object):
                       None,
                       'attemptid=%s' % (self.attemptId,),
                       None))
-    LOGGER.info('Retrieving %s' % (url,))
+    LOG.info('Retrieving %s' % (url,))
     try:
       data = urllib2.urlopen(url)
     except urllib2.URLError:
@@ -485,7 +485,7 @@ class TaskAttempt(object):
     log_sections = et.findall('body/pre')
     logs = [section.text or '' for section in log_sections]
     if len(logs) < 3:
-      LOGGER.warn('Error parsing task attempt log for %s at "%s". Found %d (not 3) log sections' %
+      LOG.warn('Error parsing task attempt log for %s at "%s". Found %d (not 3) log sections' %
                   (self.attemptId, url, len(log_sections)))
       err = _("Hue encountered an error while retrieving logs from '%s'.") % (url,)
       logs += [err] * (3 - len(logs))
@@ -582,7 +582,7 @@ class LinkJobLogs(object):
     try:
       return '<a href="%s" target="_blank">%s</a>' % (location_to_url(match.group(0), strict=False), match.group(0))
     except:
-      LOGGER.exception('failed to replace hdfs links: %s' % (match.groups(),))
+      LOG.exception('failed to replace hdfs links: %s' % (match.groups(),))
       return match.group(0)
 
   @classmethod
@@ -590,7 +590,7 @@ class LinkJobLogs(object):
     try:
       return '<a href="%s" target="_blank">%s</a>' % (reverse('jobbrowser.views.single_job', kwargs={'job': match.group(0)}), match.group(0))
     except:
-      LOGGER.exception('failed to replace mr links: %s' % (match.groups(),))
+      LOG.exception('failed to replace mr links: %s' % (match.groups(),))
       return match.group(0)
 
 

+ 14 - 8
apps/jobbrowser/src/jobbrowser/tests.py

@@ -18,6 +18,7 @@
 
 import json
 import logging
+import re
 import time
 import unittest
 
@@ -342,15 +343,20 @@ class TestJobBrowserWithHadoop(unittest.TestCase, OozieServerProvider):
     response = TestJobBrowserWithHadoop.client.get('/jobbrowser/jobs/%s/tasks?tasktext=clean' % (TestJobBrowserWithHadoop.hadoop_job_id,))
     assert_true(len(response.context['page'].object_list), 1)
 
-  def test_job_single_logs_page(self):
-    raise SkipTest
+  def test_job_single_logs(self):
+    response = TestJobBrowserWithHadoop.client.get('/jobbrowser/jobs/%s/single_logs?format=json' % (TestJobBrowserWithHadoop.hadoop_job_id))
+    json_resp = json.loads(response.content)
+
+    assert_true('logs' in json_resp)
+    assert_true('Log Type: stdout' in json_resp['logs'][1])
+    assert_true('Log Type: stderr' in json_resp['logs'][2])
+    assert_true('Log Type: syslog' in json_resp['logs'][3])
 
-    response = TestJobBrowserWithHadoop.client.get('/jobbrowser/jobs/%s/single_logs' % (TestJobBrowserWithHadoop.hadoop_job_id))
-    assert_true('syslog' in response.content, response.content)
-    assert_true('<div class="tab-pane active" id="logsSysLog">' in response.content or
-                '<div class="tab-pane active" id="logsStdErr">' in response.content or # Depending on Hadoop
-                '<div class="tab-pane active" id="logsStdOut">' in response.content, # For jenkins
-                response.content)
+    # Verify that syslog contains log information for a completed oozie job
+    match = re.search(r"^Log Type: syslog(.+)Log Length: (?P<log_length>\d+)(.+)$", json_resp['logs'][3], re.DOTALL)
+    assert_true(match and match.group(2), 'Failed to parse log length from syslog')
+    log_length = match.group(2)
+    assert_true(log_length > 0, 'Log Length is 0, expected content in syslog.')
 
 
 class TestMapReduce1NoHadoop:

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

@@ -50,7 +50,7 @@ from jobbrowser.yarn_models import Application
 import urllib2
 
 
-LOGGER = logging.getLogger(__name__)
+LOG = logging.getLogger(__name__)
 
 
 def check_job_permission(view_func):
@@ -73,7 +73,7 @@ def check_job_permission(view_func):
       raise PopupException(_('Job %s has expired.') % jobid, detail=_('Cannot be found on the History Server.'))
     except Exception, e:
       msg = 'Could not find job %s.'
-      LOGGER.exception(msg % jobid)
+      LOG.exception(msg % jobid)
       raise PopupException(_(msg) % jobid, detail=e)
 
     if not SHARE_JOBS.get() and not request.user.is_superuser \
@@ -245,7 +245,7 @@ def kill_job(request, job):
   try:
     job.kill()
   except Exception, e:
-    LOGGER.exception('Killing job')
+    LOG.exception('Killing job')
     raise PopupException(e)
 
   cur_time = time.time()
@@ -265,6 +265,7 @@ def kill_job(request, job):
 
   raise Exception(_("Job did not appear as killed within 15 seconds."))
 
+
 @check_job_permission
 def job_attempt_logs(request, job, attempt_index=0):
   return render("job_attempt_logs.mako", request, {
@@ -299,9 +300,9 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=0
     try:
       debug_info = '\nLog Link: %s' % log_link
       debug_info += '\nHTML Response: %s' % response
-      LOGGER.error(debug_info)
+      LOG.error(debug_info)
     except:
-      LOGGER.exception('failed to create debug info')
+      LOG.exception('failed to create debug info')
 
   response = {'log': LinkJobLogs._make_hdfs_links(log), 'debug': debug_info}
 

+ 37 - 6
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -365,7 +365,6 @@ class Attempt:
       for key, value in attrs.iteritems():
         setattr(self, key, value)
     self.is_mr2 = True
-
     self._fixup()
 
   def _fixup(self):
@@ -395,11 +394,43 @@ class Attempt:
     logs = []
     attempt = self.task.job.job_attempts['jobAttempt'][-1]
     log_link = attempt['logsLink']
-    # Get MR task logs
-    if self.assignedContainerId:
-      log_link = log_link.replace(attempt['containerId'], self.assignedContainerId)
-    if hasattr(self, 'nodeHttpAddress'):
-      log_link = log_link.replace(attempt['nodeHttpAddress'].split(':')[0], self.nodeHttpAddress.split(':')[0])
+
+    # Generate actual task log link from logsLink url
+    if self.task.job.status in ('NEW', 'SUBMITTED', 'RUNNING'):
+      logs_path = '/node/containerlogs/'
+      node_url, tracking_path = log_link.split(logs_path)
+      container_id, user = tracking_path.strip('/').split('/')
+
+      # Replace log path tokens with actual container properties if available
+      if hasattr(self, 'nodeHttpAddress') and 'nodeId' in attempt:
+        node_url = '%s://%s:%s' % (node_url.split('://')[0], self.nodeHttpAddress.split(':')[0], attempt['nodeId'].split(':')[1])
+      container_id = self.assignedContainerId if hasattr(self, 'assignedContainerId') else container_id
+
+      log_link = '%(node_url)s/%(logs_path)s/%(container)s/%(user)s' % {
+        'node_url': node_url,
+        'logs_path': logs_path.strip('/'),
+        'container': container_id,
+        'user': user
+      }
+    else:  # Completed jobs
+      logs_path = '/jobhistory/logs/'
+      root_url, tracking_path = log_link.split(logs_path)
+      node_url, container_id, attempt_id, user = tracking_path.strip('/').split('/')
+
+      # Replace log path tokens with actual attempt properties if available
+      if hasattr(self, 'nodeHttpAddress') and 'nodeId' in attempt:
+        node_url = '%s:%s' % (self.nodeHttpAddress.split(':')[0], attempt['nodeId'].split(':')[1])
+      container_id = self.assignedContainerId if hasattr(self, 'assignedContainerId') else container_id
+      attempt_id = self.attemptId if hasattr(self, 'attemptId') else attempt_id
+
+      log_link = '%(root_url)s/%(logs_path)s/%(node)s/%(container)s/%(attempt)s/%(user)s' % {
+        'root_url': root_url,
+        'logs_path': logs_path.strip('/'),
+        'node': node_url,
+        'container': container_id,
+        'attempt': attempt_id,
+        'user': user
+      }
 
     for name in ('stdout', 'stderr', 'syslog'):
       link = '/%s/' % name