Parcourir la source

HUE-4061 [jb] Job attempt logs not appearing for running jobs

For RUNNING MR jobs, we refer to the AM API (YARN API) and grab the amContainerLogs
For completed jobs (FAILED, SUCCEEDED, KILLED), we assemble the JHS log URL.
Jenny Kim il y a 9 ans
Parent
commit
67374ba

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

@@ -17,6 +17,8 @@
 
 import logging
 
+from django.utils.translation import ugettext as _
+
 from desktop.lib.exceptions_renderable import PopupException
 from desktop.lib.paginator import Paginator
 from desktop.lib.rest.http_client import RestException
@@ -258,6 +260,21 @@ class YarnApi(JobBrowserApi):
 
     return job
 
+  def get_application(self, jobid):
+    app = None
+    app_id = jobid.replace('job', 'application')
+
+    try:
+      app = self.resource_manager_api.app(app_id)['app']
+    except RestException, e:
+      raise PopupException(_('Job %s could not be found in Resource Manager: %s') % (jobid, e), detail=e)
+    except ApplicationNotRunning, e:
+      raise PopupException(_('Application is not running: %s') % e, detail=e)
+    except Exception, e:
+      raise PopupException(_('Job %s could not be found: %s') % (jobid, e), detail=e)
+
+    return app
+
   def get_tasks(self, jobid, **filters):
     filters.pop('pagenum')
     return self.get_job(jobid).filter_tasks(**filters)

+ 40 - 22
apps/jobbrowser/src/jobbrowser/views.py

@@ -284,35 +284,53 @@ def job_attempt_logs(request, job, attempt_index=0):
 def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=0):
   """For async log retrieval as Yarn servers are very slow"""
 
+  log_link = None
+  response = {'status': -1}
+
   try:
-    attempt_index = int(attempt_index)
-    attempt = job.job_attempts['jobAttempt'][attempt_index]
-    log_link = attempt['logsLink']
-    # Reformat log link to use YARN RM, replace node addr with node ID addr
-    log_link = log_link.replace(attempt['nodeHttpAddress'], attempt['nodeId'])
+    jt = get_api(request.user, request.jt)
+    app = jt.get_application(job.jobId)
+
+    if app['applicationType'] == 'MAPREDUCE':
+      if app['finalStatus'] in ('SUCCEEDED', 'FAILED', 'KILLED'):
+        attempt_index = int(attempt_index)
+        attempt = job.job_attempts['jobAttempt'][attempt_index]
+
+        log_link = attempt['logsLink']
+        # Reformat log link to use YARN RM, replace node addr with node ID addr
+        log_link = log_link.replace(attempt['nodeHttpAddress'], attempt['nodeId'])
+      elif app['state'] == 'RUNNING':
+        log_link = app['amContainerLogs']
   except (KeyError, RestException), e:
     raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
+  except Exception, e:
+    raise Exception(_("Failed to get application for job %s: %s") % (job.jobId, e))
 
-  link = '/%s/' % name
-  params = {}
-  if offset and int(offset) >= 0:
-    params['start'] = offset
+  if log_link:
+    link = '/%s/' % name
+    params = {}
+    if offset and int(offset) >= 0:
+      params['start'] = offset
+
+    root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
+    api_resp = None
 
-  root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
-  debug_info = ''
-  try:
-    response = root.get(link, params=params)
-    log = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
-  except Exception, e:
-    log = _('Failed to retrieve log: %s' % e)
     try:
-      debug_info = '\nLog Link: %s' % log_link
-      debug_info += '\nHTML Response: %s' % response
-      LOG.error(debug_info)
-    except:
-      LOG.exception('failed to create debug info')
+      api_resp = root.get(link, params=params)
+      log = html.fromstring(api_resp, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
 
-  response = {'log': LinkJobLogs._make_hdfs_links(log), 'debug': debug_info}
+      response['status'] = 0
+      response['log'] = LinkJobLogs._make_hdfs_links(log)
+    except Exception, e:
+      response['log'] = _('Failed to retrieve log: %s' % e)
+      try:
+        debug_info = '\nLog Link: %s' % log_link
+        if api_resp:
+          debug_info += '\nHTML Response: %s' % response
+        response['debug'] = debug_info
+        LOG.error(debug_info)
+      except:
+        LOG.exception('failed to create debug info')
 
   return JsonResponse(response)
 

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

@@ -442,6 +442,7 @@ class Attempt:
       if int(offset) >= 0:
         params['start'] = offset
 
+      response = None
       try:
         log_link = re.sub('job_[^/]+', self.id, log_link)
         root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
@@ -451,7 +452,8 @@ class Attempt:
         log = _('Failed to retrieve log: %s' % e)
         try:
           debug_info = '\nLog Link: %s' % log_link
-          debug_info += '\nHTML Response: %s' % response
+          if response:
+            debug_info += '\nHTML Response: %s' % response
           LOG.error(debug_info)
         except:
           LOG.exception('failed to build debug info')