|
|
@@ -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:
|
|
|
|