job_api.py 8.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import json
  18. import logging
  19. from django.utils.translation import ugettext as _
  20. from jobbrowser.apis.base_api import Api, MockDjangoRequest
  21. from jobbrowser.views import job_attempt_logs_json, kill_job
  22. LOG = logging.getLogger(__name__)
  23. try:
  24. from jobbrowser.api import YarnApi as NativeYarnApi
  25. except Exception, e:
  26. LOG.exception('Some application are not enabled: %s' % e)
  27. class JobApi(Api):
  28. def __init__(self, user):
  29. self.user = user
  30. self.yarn_api = YarnApi(user) # TODO: actually long term move job aggregations to the frontend instead probably
  31. self.impala_api = ImpalaApi(user)
  32. self.request = None
  33. def apps(self, filters):
  34. jobs = self.yarn_api.apps()
  35. # += Impala
  36. # += Sqoop2
  37. return jobs
  38. def app(self, appid):
  39. return self._get_api(appid).app(appid)
  40. def action(self, appid, operation):
  41. return self._get_api(appid).action(operation, appid)
  42. def logs(self, appid, app_type):
  43. return self._get_api(appid).logs(appid, app_type)
  44. def profile(self, appid, app_type, app_property):
  45. return self._get_api(appid).profile(appid, app_type, app_property)
  46. def _get_api(self, appid):
  47. if appid.startswith('task_'):
  48. return YarnMapReduceTaskApi(self.user, appid)
  49. elif appid.startswith('attempt_'):
  50. return YarnMapReduceTaskAttemptApi(self.user, appid)
  51. else:
  52. return self.yarn_api # application_
  53. def _set_request(self, request):
  54. self.request = request
  55. class YarnApi(Api):
  56. """YARN, MR, Spark"""
  57. def apps(self):
  58. jobs = NativeYarnApi(self.user).get_jobs(self.user, username=self.user.username, state='all', text='')
  59. return [{
  60. 'id': app.jobId,
  61. 'name': app.name,
  62. 'type': app.applicationType,
  63. 'status': app.status,
  64. 'apiStatus': self._api_status(app.status),
  65. 'user': self.user.username,
  66. 'progress': app.progress,
  67. 'duration': 10 * 3600,
  68. 'submitted': 10 * 3600
  69. } for app in jobs]
  70. def app(self, appid):
  71. app = NativeYarnApi(self.user).get_job(jobid=appid)
  72. common = {
  73. 'id': app.jobId,
  74. 'name': app.name,
  75. 'type': app.applicationType,
  76. 'status': app.status,
  77. 'apiStatus': self._api_status(app.status),
  78. 'user': self.user.username,
  79. 'progress': app.progress,
  80. 'duration': 10 * 3600,
  81. 'submitted': 10 * 3600
  82. }
  83. if app.applicationType == 'MR2':
  84. common['type'] = 'MAPREDUCE'
  85. common['duration'] = app.duration
  86. common['durationFormatted'] = app.durationFormatted
  87. common['properties'] = {
  88. 'maps_percent_complete': app.maps_percent_complete,
  89. 'reduces_percent_complete': app.reduces_percent_complete,
  90. 'finishedMaps': app.finishedMaps,
  91. 'finishedReduces': app.finishedReduces,
  92. 'desiredMaps': app.desiredMaps,
  93. 'desiredReduces': app.desiredReduces,
  94. 'tasks': [],
  95. 'metadata': [],
  96. 'counters': []
  97. }
  98. return common
  99. def action(self, operation, appid):
  100. if operation['action'] == 'kill':
  101. return kill_job(MockDjangoRequest(self.user), job=appid)
  102. else:
  103. return {}
  104. def logs(self, appid, app_type):
  105. if app_type == 'MAPREDUCE':
  106. response = job_attempt_logs_json(MockDjangoRequest(self.user), job=appid)
  107. logs = json.loads(response.content)['log']
  108. else:
  109. logs = None
  110. return {'logs': {'default': logs}}
  111. def profile(self, appid, app_type, app_property):
  112. if app_type == 'MAPREDUCE':
  113. if app_property == 'tasks':
  114. return {
  115. 'task_list': YarnMapReduceTaskApi(self.user, appid).apps(),
  116. }
  117. elif app_property == 'metadata':
  118. return NativeYarnApi(self.user).get_job(jobid=appid).full_job_conf
  119. elif app_property == 'counters':
  120. return NativeYarnApi(self.user).get_job(jobid=appid).counters
  121. return {}
  122. def _api_status(self, status):
  123. if status in ['NEW', 'NEW_SAVING', 'SUBMITTED', 'ACCEPTED', 'RUNNING']:
  124. return 'RUNNING'
  125. else:
  126. return 'FINISHED' # FINISHED, FAILED, KILLED
  127. class YarnMapReduceTaskApi(Api):
  128. def __init__(self, user, app_id):
  129. Api.__init__(self, user)
  130. self.app_id = '_'.join(app_id.replace('task_', 'application_').split('_')[:3])
  131. def apps(self):
  132. return [self._massage_task(task) for task in NativeYarnApi(self.user).get_tasks(jobid=self.app_id, pagenum=1)]
  133. def app(self, appid):
  134. task = NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=appid)
  135. common = self._massage_task(task)
  136. common['properties'] = {
  137. 'attempts': [],
  138. 'metadata': [],
  139. 'counters': []
  140. }
  141. common['properties'].update(self._massage_task(task))
  142. return common
  143. def logs(self, appid, app_type):
  144. response = job_attempt_logs_json(MockDjangoRequest(self.user), job=self.app_id)
  145. logs = json.loads(response.content)['log']
  146. return {'progress': 0, 'logs': {'default': logs}}
  147. def profile(self, appid, app_type, app_property):
  148. if app_property == 'attempts':
  149. return {
  150. 'task_list': YarnMapReduceTaskAttemptApi(self.user, appid).apps(),
  151. }
  152. elif app_property == 'counters':
  153. return NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=appid).counters
  154. return {}
  155. def _massage_task(self, task):
  156. return {
  157. 'id': task.id,
  158. 'type': task.type,
  159. 'elapsedTime': task.elapsedTime,
  160. 'progress': task.progress,
  161. 'state': task.state,
  162. 'startTime': task.startTime,
  163. 'successfulAttempt': task.successfulAttempt,
  164. 'finishTime': task.finishTime
  165. }
  166. class YarnMapReduceTaskAttemptApi(Api):
  167. def __init__(self, user, app_id):
  168. Api.__init__(self, user)
  169. self.app_id = '_'.join(app_id.replace('task_', 'application_').replace('attempt_', 'application_').split('_')[:3])
  170. self.task_id = '_'.join(app_id.replace('attempt_', 'task_').split('_')[:5])
  171. self.attempt_id = app_id
  172. def apps(self):
  173. return [self._massage_task(task) for task in NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=self.task_id).attempts]
  174. def app(self, appid):
  175. task = NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=self.task_id).get_attempt(self.attempt_id)
  176. common = self._massage_task(task)
  177. common['properties'] = {
  178. 'metadata': [],
  179. 'counters': []
  180. }
  181. common['properties'].update(self._massage_task(task))
  182. return common
  183. def logs(self, appid, app_type):
  184. task = NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=self.task_id).get_attempt(self.attempt_id)
  185. stdout, stderr, syslog = task.get_task_log()
  186. return {'progress': 0, 'logs': {'default': stdout, 'stdout': stdout, 'stderr': stderr, 'syslog': syslog}}
  187. def profile(self, appid, app_type, app_property):
  188. if app_property == 'counters':
  189. return NativeYarnApi(self.user).get_task(jobid=self.app_id, task_id=self.task_id).get_attempt(self.attempt_id).counters
  190. return {}
  191. def _massage_task(self, task):
  192. return {
  193. #"elapsedMergeTime" : task.elapsedMergeTime,
  194. #"shuffleFinishTime" : task.shuffleFinishTime,
  195. "assignedContainerId" : task.assignedContainerId,
  196. "progress" : task.progress,
  197. "elapsedTime" : task.elapsedTime,
  198. "state" : task.state,
  199. #"elapsedShuffleTime" : task.elapsedShuffleTime,
  200. #"mergeFinishTime" : task.mergeFinishTime,
  201. "rack" : task.rack,
  202. #"elapsedReduceTime" : task.elapsedReduceTime,
  203. "nodeHttpAddress" : task.nodeHttpAddress,
  204. "type" : task.type + '_ATTEMPT',
  205. "startTime" : task.startTime,
  206. "id" : task.id,
  207. "finishTime" : task.finishTime
  208. }
  209. class YarnAtsApi(Api):
  210. pass
  211. class ImpalaApi(Api):
  212. pass
  213. class Sqoop2Api(Api):
  214. pass