浏览代码

HUE-8677 [oozie] Oozie batch snippet hangs (#787)

* HUE-8677 [oozie] Remove usage of appAttemptId

* HUE-8677 [oozie] Fix appattempts_attempt RM API call

* HUE-8677 [oozie] Remove usage of non-existent diagnosticsInfo field

* HUE-8677 [oozie] Fix 'object of type 'int' has no len()'

* HUE-8677 [oozie] Handle case when logsLink points to NM instead of JHS for comleted jobs
Ivan Dzikovsky 6 年之前
父节点
当前提交
4a0b2e9057

+ 2 - 2
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -898,7 +898,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             <tr>
               <th>${_('Assigned Container Id')}</th>
               <th>${_('Node Id')}</th>
-              <th>${_('appAttemptId')}</th>
+              <th>${_('Application Attempt Id')}</th>
               <th>${_('Start Time')}</th>
               <th>${_('Finish Time')}</th>
               <th>${_('Node Http Address')}</th>
@@ -910,7 +910,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
               <tr class="pointer" data-bind="click: function() { $root.job().id(id); $root.job().fetchJob(); }">
                 <td data-bind="text: containerId"></td>
                 <td data-bind="text: nodeId"></td>
-                <td data-bind="text: appAttemptId"></td>
+                <td data-bind="text: id"></td>
                 <td data-bind="moment: {data: startTime, format: 'LLL'}"></td>
                 <td data-bind="moment: {data: finishedTime, format: 'LLL'}"></td>
                 <td data-bind="text: nodeHttpAddress"></td>

+ 28 - 17
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -393,7 +393,7 @@ class OozieYarnJob(Job):
   def get_task(self, task_id):
     task = YarnTask(self)
     task.taskId = None
-    task.taskAttemptIds = [appAttempt['appAttemptId'] for appAttempt in self.job_attempts['jobAttempt']]
+    task.taskAttemptIds = [appAttempt['id'] for appAttempt in self.job_attempts['jobAttempt']]
     return task
 
   def filter_tasks(self, task_types=None, task_states=None, task_text=None):
@@ -403,8 +403,6 @@ class OozieYarnJob(Job):
   def job_attempts(self):
     if not hasattr(self, '_job_attempts'):
       attempts = self.api.appattempts(self.id)['appAttempts']['appAttempt']
-      for attempt in attempts:
-        attempt['id'] = attempt['appAttemptId']
       self._job_attempts = {
         'jobAttempt': attempts
       }
@@ -547,7 +545,7 @@ class Attempt:
     log_link = attempt['logsLink']
 
     # Generate actual task log link from logsLink url
-    if self.task.job.status in ('NEW', 'SUBMITTED', 'RUNNING') or self.type == 'Oozie Launcher':
+    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('/')
@@ -564,19 +562,32 @@ class Attempt:
         '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('/')
+      if '/node/containerlogs/' in log_link:
+        # Applications that use NodeManager API instead of JobHistory API, like new "Oozie Launcher",
+        # have `logsLink` URL pointed to NodeManager even for completed jobs
+        logs_path = '/node/containerlogs/'
 
-      # 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
+        yarn_cluster = cluster.get_cluster_conf_for_job_submission()
+        root_url = yarn_cluster.HISTORY_SERVER_API_URL.get()
+
+        tracking_path = log_link.split(logs_path)[1]
+        container_id, user = tracking_path.strip('/').split('/')
+
+        node_url = self.nodeId
+        attempt_id = self.id
+      else:
+        logs_path = '/jobhistory/logs/'
+        root_url, tracking_path = log_link.split(logs_path)
+        node_url, container_id, attempt_id, user = tracking_path.strip('/').split('/')
 
-      log_link = '%(root_url)s/%(logs_path)s/%(node)s/%(container)s/%(attempt)s/%(user)s' % {
+        # 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/jobhistory/logs/%(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,
@@ -599,7 +610,7 @@ class Attempt:
 
       response = None
       try:
-        log_link = re.sub('job_[^/]+', self.id, log_link)
+        log_link = re.sub('job_[^/]+', str(self.id), log_link)
         root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
         response = root.get(link, params=params)
         log = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
@@ -627,9 +638,9 @@ class YarnOozieAttempt(Attempt):
     self._fixup()
 
   def _fixup(self):
-    setattr(self, 'diagnostics', self.diagnosticsInfo)
+    if not hasattr(self, 'diagnostics'):
+      self.diagnostics = ''
     setattr(self, 'type', 'Oozie Launcher')
-    setattr(self, 'id', self.appAttemptId)
 
 class Container:
 

+ 5 - 2
desktop/libs/hadoop/src/hadoop/yarn/resource_manager_api.py

@@ -129,8 +129,11 @@ class ResourceManagerApi(object):
     return self._execute(self._root.get, 'cluster/apps/%(app_id)s/appattempts' % {'app_id': app_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
 
   def appattempts_attempt(self, app_id, attempt_id):
-    params = self._get_params()
-    return self._execute(self._root.get, 'cluster/apps/%(app_id)s/appattempts/%(attempt_id)s' % {'app_id': app_id, 'attempt_id': attempt_id}, params=params, headers={'Accept': _JSON_CONTENT_TYPE})
+    attempts = self.appattempts(app_id)
+    for attempt in attempts['appAttempts']['appAttempt']:
+      if attempt['id'] == attempt_id:
+        return attempt
+    raise PopupException('Application {} does not have application attempt with id {}'.format(app_id, attempt_id))
 
   def kill(self, app_id):
     data = {'state': 'KILLED'}