Sfoglia il codice sorgente

HUE-8737 [core] Futurize apps/jobbrowser for Python 3.5

Ying Chen 6 anni fa
parent
commit
5b314775d4

+ 13 - 14
apps/jobbrowser/src/jobbrowser/api.py

@@ -15,6 +15,7 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from builtins import object
 import logging
 import logging
 
 
 from datetime import datetime, timedelta
 from datetime import datetime, timedelta
@@ -73,7 +74,7 @@ class YarnApi(JobBrowserApi):
     self.mapreduce_api = None
     self.mapreduce_api = None
     self.history_server_api = None
     self.history_server_api = None
     self.spark_history_server_api = None
     self.spark_history_server_api = None
-    if YARN_CLUSTERS.keys():
+    if list(YARN_CLUSTERS.keys()):
       self.resource_manager_api = resource_manager_api.get_resource_manager(user.username)
       self.resource_manager_api = resource_manager_api.get_resource_manager(user.username)
       self.mapreduce_api = mapreduce_api.get_mapreduce_api(user.username)
       self.mapreduce_api = mapreduce_api.get_mapreduce_api(user.username)
       self.history_server_api = history_server_api.get_history_server_api(user.username)
       self.history_server_api = history_server_api.get_history_server_api(user.username)
@@ -113,11 +114,10 @@ class YarnApi(JobBrowserApi):
 
 
     if kwargs['text']:
     if kwargs['text']:
       text = kwargs['text'].lower()
       text = kwargs['text'].lower()
-      jobs = filter(lambda job:
-                    text in job.name.lower() or
+      jobs = [job for job in jobs if text in job.name.lower() or
                     text in job.id.lower() or
                     text in job.id.lower() or
                     text in job.user.lower() or
                     text in job.user.lower() or
-                    text in job.queue.lower(), jobs)
+                    text in job.queue.lower()]
 
 
     return self.filter_jobs(user, jobs)
     return self.filter_jobs(user, jobs)
 
 
@@ -135,10 +135,9 @@ class YarnApi(JobBrowserApi):
   def filter_jobs(self, user, jobs, **kwargs):
   def filter_jobs(self, user, jobs, **kwargs):
     check_permission = not SHARE_JOBS.get() and not is_admin(user)
     check_permission = not SHARE_JOBS.get() and not is_admin(user)
 
 
-    return filter(lambda job:
-                  not check_permission or
+    return [job for job in jobs if not check_permission or
                   is_admin(user) or
                   is_admin(user) or
-                  job.user == user.username, jobs)
+                  job.user == user.username]
 
 
 
 
   def _get_job_from_history_server(self, job_id):
   def _get_job_from_history_server(self, job_id):
@@ -175,20 +174,20 @@ class YarnApi(JobBrowserApi):
           job = SparkJob(app, rm_api=self.resource_manager_api, hs_api=self.spark_history_server_api)
           job = SparkJob(app, rm_api=self.resource_manager_api, hs_api=self.spark_history_server_api)
         else:
         else:
           job = Application(app, self.resource_manager_api)
           job = Application(app, self.resource_manager_api)
-    except RestException, e:
+    except RestException as e:
       if e.code == 404:  # Job not found in RM so attempt to find job in JHS
       if e.code == 404:  # Job not found in RM so attempt to find job in JHS
         job = self._get_job_from_history_server(job_id)
         job = self._get_job_from_history_server(job_id)
       else:
       else:
         LOG.error("Job %s has expired: %s" % (app_id, e))
         LOG.error("Job %s has expired: %s" % (app_id, e))
         raise JobExpired(app_id)
         raise JobExpired(app_id)
-    except PopupException, e:
+    except PopupException as e:
       if 'NotFoundException' in e.message:
       if 'NotFoundException' in e.message:
         job = self._get_job_from_history_server(job_id)
         job = self._get_job_from_history_server(job_id)
       else:
       else:
         raise e
         raise e
-    except ApplicationNotRunning, e:
+    except ApplicationNotRunning as e:
       raise e
       raise e
-    except Exception, e:
+    except Exception as e:
       raise PopupException('Job %s could not be found: %s' % (jobid, e), detail=e)
       raise PopupException('Job %s could not be found: %s' % (jobid, e), detail=e)
 
 
     return job
     return job
@@ -199,11 +198,11 @@ class YarnApi(JobBrowserApi):
 
 
     try:
     try:
       app = self.resource_manager_api.app(app_id)['app']
       app = self.resource_manager_api.app(app_id)['app']
-    except RestException, e:
+    except RestException as e:
       raise PopupException(_('Job %s could not be found in Resource Manager: %s') % (jobid, e), detail=e)
       raise PopupException(_('Job %s could not be found in Resource Manager: %s') % (jobid, e), detail=e)
-    except ApplicationNotRunning, e:
+    except ApplicationNotRunning as e:
       raise PopupException(_('Application is not running: %s') % e, detail=e)
       raise PopupException(_('Application is not running: %s') % e, detail=e)
-    except Exception, e:
+    except Exception as e:
       raise PopupException(_('Job %s could not be found: %s') % (jobid, e), detail=e)
       raise PopupException(_('Job %s could not be found: %s') % (jobid, e), detail=e)
 
 
     return app
     return app

+ 3 - 3
apps/jobbrowser/src/jobbrowser/api2.py

@@ -36,7 +36,7 @@ def api_error_handler(func):
 
 
     try:
     try:
       return func(*args, **kwargs)
       return func(*args, **kwargs)
-    except Exception, e:
+    except Exception as e:
       LOG.exception('Error running %s' % func)
       LOG.exception('Error running %s' % func)
       response['status'] = -1
       response['status'] = -1
       response['message'] = smart_unicode(e)
       response['message'] = smart_unicode(e)
@@ -53,7 +53,7 @@ def jobs(request, interface=None):
 
 
   cluster = json.loads(request.POST.get('cluster', '{}'))
   cluster = json.loads(request.POST.get('cluster', '{}'))
   interface = json.loads(request.POST.get('interface'))
   interface = json.loads(request.POST.get('interface'))
-  filters = dict([(key, value) for _filter in json.loads(request.POST.get('filters', '[]')) for key, value in _filter.items() if value])
+  filters = dict([(key, value) for _filter in json.loads(request.POST.get('filters', '[]')) for key, value in list(_filter.items()) if value])
 
 
   jobs = get_api(request.user, interface, cluster=cluster).apps(filters)
   jobs = get_api(request.user, interface, cluster=cluster).apps(filters)
 
 
@@ -126,7 +126,7 @@ def profile(request):
   app_id = json.loads(request.POST.get('app_id'))
   app_id = json.loads(request.POST.get('app_id'))
   app_type = json.loads(request.POST.get('app_type'))
   app_type = json.loads(request.POST.get('app_type'))
   app_property = json.loads(request.POST.get('app_property'))
   app_property = json.loads(request.POST.get('app_property'))
-  app_filters = dict([(key, value) for _filter in json.loads(request.POST.get('app_filters', '[]')) for key, value in _filter.items() if value])
+  app_filters = dict([(key, value) for _filter in json.loads(request.POST.get('app_filters', '[]')) for key, value in list(_filter.items()) if value])
 
 
   api = get_api(request.user, interface, cluster=cluster)
   api = get_api(request.user, interface, cluster=cluster)
   api._set_request(request) # For YARN
   api._set_request(request) # For YARN

+ 3 - 2
apps/jobbrowser/src/jobbrowser/apis/base_api.py

@@ -15,6 +15,7 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from builtins import object
 import logging
 import logging
 import posixpath
 import posixpath
 import re
 import re
@@ -89,7 +90,7 @@ class Api(object):
     self.request = request
     self.request = request
 
 
 
 
-class MockDjangoRequest():
+class MockDjangoRequest(object):
 
 
   def __init__(self, user, get=None, post=None, method='POST'):
   def __init__(self, user, get=None, post=None, method='POST'):
     self.user = user
     self.user = user
@@ -103,7 +104,7 @@ class MockDjangoRequest():
 def _extract_query_params(filters):
 def _extract_query_params(filters):
   filter_params = {}
   filter_params = {}
 
 
-  for name, value in filters.iteritems():
+  for name, value in filters.items():
     if name == 'text':
     if name == 'text':
       filter_params['text'] = value
       filter_params['text'] = value
       user_filter = re.search('((user):([^ ]+))', value)
       user_filter = re.search('((user):([^ ]+))', value)

+ 1 - 1
apps/jobbrowser/src/jobbrowser/apis/bundle_api.py

@@ -33,7 +33,7 @@ LOG = logging.getLogger(__name__)
 try:
 try:
   from oozie.conf import OOZIE_JOBS_COUNT
   from oozie.conf import OOZIE_JOBS_COUNT
   from oozie.views.dashboard import get_oozie_job_log, massaged_oozie_jobs_for_json, list_oozie_bundle
   from oozie.views.dashboard import get_oozie_job_log, massaged_oozie_jobs_for_json, list_oozie_bundle
-except Exception, e:
+except Exception as e:
   LOG.exception('Some application are not enabled: %s' % e)
   LOG.exception('Some application are not enabled: %s' % e)
 
 
 
 

+ 10 - 10
apps/jobbrowser/src/jobbrowser/apis/job_api.py

@@ -38,7 +38,7 @@ try:
   from jobbrowser.apis.base_api import Api, MockDjangoRequest, _extract_query_params
   from jobbrowser.apis.base_api import Api, MockDjangoRequest, _extract_query_params
   from jobbrowser.views import job_attempt_logs_json, kill_job, massage_job_for_json
   from jobbrowser.views import job_attempt_logs_json, kill_job, massage_job_for_json
   from jobbrowser.yarn_models import Application
   from jobbrowser.yarn_models import Application
-except Exception, e:
+except Exception as e:
   LOG.exception('Some application are not enabled: %s' % e)
   LOG.exception('Some application are not enabled: %s' % e)
 
 
 
 
@@ -130,15 +130,15 @@ class YarnApi(Api):
   def app(self, appid):
   def app(self, appid):
     try:
     try:
       job = NativeYarnApi(self.user).get_job(jobid=appid)
       job = NativeYarnApi(self.user).get_job(jobid=appid)
-    except ApplicationNotRunning, e:
+    except ApplicationNotRunning as e:
       if e.job.get('state', '').lower() == 'accepted':
       if e.job.get('state', '').lower() == 'accepted':
         rm_api = resource_manager_api.get_resource_manager(self.user)
         rm_api = resource_manager_api.get_resource_manager(self.user)
         job = Application(e.job, rm_api)
         job = Application(e.job, rm_api)
       else:
       else:
         raise e  # Job has not yet been accepted by RM
         raise e  # Job has not yet been accepted by RM
-    except JobExpired, e:
+    except JobExpired as e:
       raise PopupException(_('Job %s has expired.') % appid, detail=_('Cannot be found on the History Server.'))
       raise PopupException(_('Job %s has expired.') % appid, detail=_('Cannot be found on the History Server.'))
-    except Exception, e:
+    except Exception as e:
       msg = 'Could not find job %s.'
       msg = 'Could not find job %s.'
       LOG.exception(msg % appid)
       LOG.exception(msg % appid)
       raise PopupException(_(msg) % appid, detail=e)
       raise PopupException(_(msg) % appid, detail=e)
@@ -186,7 +186,7 @@ class YarnApi(Api):
       app['trackingUrl'] = job.trackingUrl if hasattr(job, 'trackingUrl') else ''
       app['trackingUrl'] = job.trackingUrl if hasattr(job, 'trackingUrl') else ''
       common['type'] = 'SPARK'
       common['type'] = 'SPARK'
       common['properties'] = {
       common['properties'] = {
-        'metadata': [{'name': name, 'value': value} for name, value in app.iteritems() if name != "url" and name != "killUrl"],
+        'metadata': [{'name': name, 'value': value} for name, value in app.items() if name != "url" and name != "killUrl"],
         'executors': []
         'executors': []
       }
       }
       if hasattr(job, 'metrics'):
       if hasattr(job, 'metrics'):
@@ -242,7 +242,7 @@ class YarnApi(Api):
         logs = json.loads(response.content).get('log')
         logs = json.loads(response.content).get('log')
       else:
       else:
         logs = None
         logs = None
-    except PopupException, e:
+    except PopupException as e:
       LOG.warn('No task attempt found for logs: %s' % smart_str(e))
       LOG.warn('No task attempt found for logs: %s' % smart_str(e))
     return {'logs': logs, 'logsList': logs_list}
     return {'logs': logs, 'logsList': logs_list}
 
 
@@ -339,8 +339,8 @@ class YarnAttemptApi(Api):
         'containerId' : task.containerId if hasattr(task, 'containerId') else '',
         'containerId' : task.containerId if hasattr(task, 'containerId') else '',
         'diagnostics': task.diagnostics if hasattr(task, 'diagnostics') else '',
         'diagnostics': task.diagnostics if hasattr(task, 'diagnostics') else '',
         "startTimeFormatted" : task.startTimeFormatted if hasattr(task, 'startTimeFormatted') else '',
         "startTimeFormatted" : task.startTimeFormatted if hasattr(task, 'startTimeFormatted') else '',
-        "startTime" : long(task.startTime) if hasattr(task, 'startTime') else '',
-        "finishTime" : long(task.finishedTime) if hasattr(task, 'finishedTime') else '',
+        "startTime" : int(task.startTime) if hasattr(task, 'startTime') else '',
+        "finishTime" : int(task.finishedTime) if hasattr(task, 'finishedTime') else '',
         "finishTimeFormatted" : task.finishTimeFormatted if hasattr(task, 'finishTimeFormatted') else '',
         "finishTimeFormatted" : task.finishTimeFormatted if hasattr(task, 'finishTimeFormatted') else '',
         "type" : task.type + '_ATTEMPT' if hasattr(task, 'type') else '',
         "type" : task.type + '_ATTEMPT' if hasattr(task, 'type') else '',
         'nodesBlacklistedBySystem': task.nodesBlacklistedBySystem if hasattr(task, 'nodesBlacklistedBySystem') else '',
         'nodesBlacklistedBySystem': task.nodesBlacklistedBySystem if hasattr(task, 'nodesBlacklistedBySystem') else '',
@@ -417,7 +417,7 @@ class YarnMapReduceTaskApi(Api):
     try:
     try:
       response = job_attempt_logs_json(MockDjangoRequest(self.user), job=self.app_id, name=log_name, is_embeddable=is_embeddable)
       response = job_attempt_logs_json(MockDjangoRequest(self.user), job=self.app_id, name=log_name, is_embeddable=is_embeddable)
       logs = json.loads(response.content)['log']
       logs = json.loads(response.content)['log']
-    except PopupException, e:
+    except PopupException as e:
       LOG.warn('No task attempt found for default logs: %s' % e)
       LOG.warn('No task attempt found for default logs: %s' % e)
       logs = ''
       logs = ''
     return {'progress': 0, 'logs': logs}
     return {'progress': 0, 'logs': logs}
@@ -530,7 +530,7 @@ class YarnMapReduceTaskAttemptApi(Api):
         "type" : task.type + '_ATTEMPT' if hasattr(task, 'type') else '',
         "type" : task.type + '_ATTEMPT' if hasattr(task, 'type') else '',
         "startTime" : task.startTime if hasattr(task, 'startTime') else '',
         "startTime" : task.startTime if hasattr(task, 'startTime') else '',
         "id" : task.id if hasattr(task, 'id') else task.appAttemptId if hasattr(task, 'appAttemptId') else '',
         "id" : task.id if hasattr(task, 'id') else task.appAttemptId if hasattr(task, 'appAttemptId') else '',
-        "finishTime" : task.finishTime if hasattr(task, 'finishTime') else long(task.finishedTime) if hasattr(task, 'finishedTime') else '',
+        "finishTime" : task.finishTime if hasattr(task, 'finishTime') else int(task.finishedTime) if hasattr(task, 'finishedTime') else '',
         "app_id": self.app_id,
         "app_id": self.app_id,
         "task_id": self.task_id,
         "task_id": self.task_id,
         'apiStatus': self._api_status(task.state) if hasattr(task, 'state') else self._api_status(task.appAttemptState) if hasattr(task, 'appAttemptState') else '',
         'apiStatus': self._api_status(task.state) if hasattr(task, 'state') else self._api_status(task.appAttemptState) if hasattr(task, 'appAttemptState') else '',

+ 6 - 4
apps/jobbrowser/src/jobbrowser/apis/query_api.py

@@ -15,6 +15,8 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from builtins import filter
+from builtins import range
 import itertools
 import itertools
 import logging
 import logging
 import re
 import re
@@ -34,7 +36,7 @@ try:
   from beeswax.models import Session
   from beeswax.models import Session
   from impala.server import get_api as get_impalad_api, _get_impala_server_url
   from impala.server import get_api as get_impalad_api, _get_impala_server_url
   from impala.dbms import _get_server_name
   from impala.dbms import _get_server_name
-except Exception, e:
+except Exception as e:
   LOG.exception('Some application are not enabled: %s' % e)
   LOG.exception('Some application are not enabled: %s' % e)
 
 
 
 
@@ -196,7 +198,7 @@ class QueryApi(Api):
     query['plan'] = query.get('plan').strip() if query.get('plan') else ''
     query['plan'] = query.get('plan').strip() if query.get('plan') else ''
     try:
     try:
       query['metrics'] = self._metrics(appid)
       query['metrics'] = self._metrics(appid)
-    except Exception, e:
+    except Exception as e:
       query['metrics'] = {'nodes' : {}}
       query['metrics'] = {'nodes' : {}}
       LOG.exception('Could not parse profile: %s' % e)
       LOG.exception('Could not parse profile: %s' % e)
 
 
@@ -289,7 +291,7 @@ class QueryApi(Api):
       def make_lambda(name, value):
       def make_lambda(name, value):
         return lambda app: app[name] == value
         return lambda app: app[name] == value
 
 
-      for key, name in filter_names.items():
+      for key, name in list(filter_names.items()):
           text_filter = re.search(r"\s*("+key+")\s*:([^ ]+)", filters.get("text"))
           text_filter = re.search(r"\s*("+key+")\s*:([^ ]+)", filters.get("text"))
           if text_filter and text_filter.group(1) == key:
           if text_filter and text_filter.group(1) == key:
             filter_list.append(make_lambda(name, text_filter.group(2).strip()))
             filter_list.append(make_lambda(name, text_filter.group(2).strip()))
@@ -305,5 +307,5 @@ class QueryApi(Api):
 
 
   def _n_filter(self, filters, tuples):
   def _n_filter(self, filters, tuples):
     for f in filters:
     for f in filters:
-      tuples = filter(f, tuples)
+      tuples = list(filter(f, tuples))
     return tuples
     return tuples

+ 3 - 2
apps/jobbrowser/src/jobbrowser/apis/schedule_api.py

@@ -15,6 +15,7 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from builtins import object
 import logging
 import logging
 import json
 import json
 
 
@@ -33,7 +34,7 @@ LOG = logging.getLogger(__name__)
 try:
 try:
   from oozie.conf import OOZIE_JOBS_COUNT
   from oozie.conf import OOZIE_JOBS_COUNT
   from oozie.views.dashboard import list_oozie_coordinator, get_oozie_job_log, massaged_oozie_jobs_for_json, has_job_edition_permission
   from oozie.views.dashboard import list_oozie_coordinator, get_oozie_job_log, massaged_oozie_jobs_for_json, has_job_edition_permission
-except Exception, e:
+except Exception as e:
   LOG.warn('Some application are not enabled: %s' % e)
   LOG.warn('Some application are not enabled: %s' % e)
 
 
 
 
@@ -162,7 +163,7 @@ class ScheduleApi(Api):
     return self._TASK_API_STATUSES.get(status, 'FAILED')
     return self._TASK_API_STATUSES.get(status, 'FAILED')
 
 
 
 
-class MockGet():
+class MockGet(object):
   def __ini__(self, statuses):
   def __ini__(self, statuses):
     self.statuses = []
     self.statuses = []
 
 

+ 4 - 4
apps/jobbrowser/src/jobbrowser/apis/workflow_api.py

@@ -33,7 +33,7 @@ try:
   from oozie.views.dashboard import get_oozie_job_log, list_oozie_workflow, manage_oozie_jobs, bulk_manage_oozie_jobs, has_dashboard_jobs_access, massaged_oozie_jobs_for_json, \
   from oozie.views.dashboard import get_oozie_job_log, list_oozie_workflow, manage_oozie_jobs, bulk_manage_oozie_jobs, has_dashboard_jobs_access, massaged_oozie_jobs_for_json, \
       has_job_edition_permission
       has_job_edition_permission
   has_oozie_installed = True
   has_oozie_installed = True
-except Exception, e:
+except Exception as e:
   LOG.warn('Some applications are not enabled for Job Browser v2: %s' % e)
   LOG.warn('Some applications are not enabled for Job Browser v2: %s' % e)
   has_oozie_installed = False
   has_oozie_installed = False
 
 
@@ -128,7 +128,7 @@ class WorkflowApi(Api):
       workflow = oozie_api.get_job(jobid=appid)
       workflow = oozie_api.get_job(jobid=appid)
       return {
       return {
         'properties': workflow.conf_dict,
         'properties': workflow.conf_dict,
-        'properties_display': [{'name': key, 'value': val, 'link': is_linkable(key, val) and hdfs_link_js(val)} for key, val in workflow.conf_dict.iteritems()],
+        'properties_display': [{'name': key, 'value': val, 'link': is_linkable(key, val) and hdfs_link_js(val)} for key, val in workflow.conf_dict.items()],
       }
       }
 
 
     return {}
     return {}
@@ -146,7 +146,7 @@ class WorkflowApi(Api):
   def _get_variables(self, workflow):
   def _get_variables(self, workflow):
     parameters = []
     parameters = []
 
 
-    for var, val in workflow.conf_dict.iteritems():
+    for var, val in workflow.conf_dict.items():
       if var not in ParameterForm.NON_PARAMETERS and var != 'oozie.use.system.libpath' or var == 'oozie.wf.application.path':
       if var not in ParameterForm.NON_PARAMETERS and var != 'oozie.use.system.libpath' or var == 'oozie.wf.application.path':
         link = ''
         link = ''
         if is_linkable(var, val):
         if is_linkable(var, val):
@@ -174,7 +174,7 @@ class WorkflowActionApi(Api):
     common['properties']['conf'] = properties.pop('conf')
     common['properties']['conf'] = properties.pop('conf')
     common['properties']['externalId'] = properties.get('externalId', '')
     common['properties']['externalId'] = properties.get('externalId', '')
     common['properties']['externalChildIDs'] = properties.get('externalChildIDs') and properties.pop('externalChildIDs').split(',')
     common['properties']['externalChildIDs'] = properties.get('externalChildIDs') and properties.pop('externalChildIDs').split(',')
-    common['properties']['properties'] = [{'name': key, 'value': val} for key, val in properties.iteritems()]
+    common['properties']['properties'] = [{'name': key, 'value': val} for key, val in properties.items()]
 
 
     common['properties']['workflow_id'] = appid.split('@', 1)[0]
     common['properties']['workflow_id'] = appid.split('@', 1)[0]
 
 

+ 5 - 1
apps/jobbrowser/src/jobbrowser/models.py

@@ -15,6 +15,10 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from __future__ import division
+from builtins import str
+from past.utils import old_div
+from builtins import object
 import datetime
 import datetime
 import logging
 import logging
 import functools
 import functools
@@ -106,6 +110,6 @@ def format_unixtime_ms(unixtime):
   Format a unix timestamp in ms to a human readable string
   Format a unix timestamp in ms to a human readable string
   """
   """
   if unixtime:
   if unixtime:
-    return str(datetime.datetime.fromtimestamp(unixtime/1000).strftime("%x %X %Z"))
+    return str(datetime.datetime.fromtimestamp(old_div(unixtime,1000)).strftime("%x %X %Z"))
   else:
   else:
     return ""
     return ""

+ 3 - 1
apps/jobbrowser/src/jobbrowser/templatetags/unix_ms_to_datetime.py

@@ -15,6 +15,8 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from __future__ import division
+from past.utils import old_div
 import datetime
 import datetime
 import django
 import django
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
@@ -25,7 +27,7 @@ register = django.template.Library()
 def unix_ms_to_datetime(unixtime):
 def unix_ms_to_datetime(unixtime):
   """unixtime is seconds since the epoch"""
   """unixtime is seconds since the epoch"""
   if unixtime:
   if unixtime:
-    return datetime.datetime.fromtimestamp(unixtime/1000)
+    return datetime.datetime.fromtimestamp(old_div(unixtime,1000))
   return _("No time")
   return _("No time")
 unix_ms_to_datetime.is_safe = True
 unix_ms_to_datetime.is_safe = True
 
 

+ 12 - 10
apps/jobbrowser/src/jobbrowser/tests.py

@@ -16,6 +16,8 @@
 # limitations under the License.
 # limitations under the License.
 
 
 
 
+from builtins import range
+from builtins import object
 import json
 import json
 import logging
 import logging
 import re
 import re
@@ -51,7 +53,7 @@ LOG = logging.getLogger(__name__)
 _INITIALIZED = False
 _INITIALIZED = False
 
 
 
 
-class TestBrowser():
+class TestBrowser(object):
 
 
   def test_format_counter_name(self):
   def test_format_counter_name(self):
     assert_equal("Foo Bar", views.format_counter_name("fooBar"))
     assert_equal("Foo Bar", views.format_counter_name("fooBar"))
@@ -154,7 +156,7 @@ class TestJobBrowserWithHadoop(unittest.TestCase, OozieServerProvider):
 
 
     cls.client.post(reverse('oozie:install_examples'))
     cls.client.post(reverse('oozie:install_examples'))
     cls.cluster.fs.do_as_user(cls.username, cls.cluster.fs.create_home_dir, cls.home_dir)
     cls.cluster.fs.do_as_user(cls.username, cls.cluster.fs.create_home_dir, cls.home_dir)
-    cls.cluster.fs.do_as_superuser(cls.cluster.fs.chmod, cls.home_dir, 0777, True)
+    cls.cluster.fs.do_as_superuser(cls.cluster.fs.chmod, cls.home_dir, 0o777, True)
 
 
     _INITIALIZED = True
     _INITIALIZED = True
 
 
@@ -358,7 +360,7 @@ class TestJobBrowserWithHadoop(unittest.TestCase, OozieServerProvider):
     assert_true(log_length > 0, 'Log Length is 0, expected content in syslog.')
     assert_true(log_length > 0, 'Log Length is 0, expected content in syslog.')
 
 
 
 
-class TestMapReduce2NoHadoop:
+class TestMapReduce2NoHadoop(object):
 
 
   def setUp(self):
   def setUp(self):
     # Beware: Monkey patching
     # Beware: Monkey patching
@@ -491,7 +493,7 @@ class TestMapReduce2NoHadoop:
 
 
 
 
 
 
-class TestResourceManagerHaNoHadoop:
+class TestResourceManagerHaNoHadoop(object):
 
 
   def setUp(self):
   def setUp(self):
     # Beware: Monkey patching
     # Beware: Monkey patching
@@ -605,14 +607,14 @@ class TestImpalaApi(object):
     response = self.api.apps({})
     response = self.api.apps({})
     target = [{'status': u'FINISHED', 'rows_fetched': 28, 'user': u'admin', 'canWrite': False, 'duration': 3355000.0, 'id': u'8a46a8865624698f:b80b211500000000', 'apiStatus': 'SUCCEEDED', 'name': u'SELECT sample_07.description, sample_07.salary FROM   sample...', 'submitted': u'2017-10-25 15:38:26.637010000', 'queue': u'root.admin', 'waiting': True, 'progress': u'1 / 1 ( 100%)', 'type': u'QUERY', 'waiting_time': u'52m8s'}, {'status': u'FINISHED', 'rows_fetched': 53, 'user': u'admin', 'canWrite': False, 'duration': 3369000.0, 'id': u'4d497267f34ff17d:817bdfb500000000', 'apiStatus': 'SUCCEEDED', 'name': u'select * from customers', 'submitted': u'2017-10-25 15:38:12.872825000', 'queue': u'root.admin', 'waiting': True, 'progress': u'2 / 3 (66.6667%)', 'type': u'QUERY', 'waiting_time': u'52m8s'}]
     target = [{'status': u'FINISHED', 'rows_fetched': 28, 'user': u'admin', 'canWrite': False, 'duration': 3355000.0, 'id': u'8a46a8865624698f:b80b211500000000', 'apiStatus': 'SUCCEEDED', 'name': u'SELECT sample_07.description, sample_07.salary FROM   sample...', 'submitted': u'2017-10-25 15:38:26.637010000', 'queue': u'root.admin', 'waiting': True, 'progress': u'1 / 1 ( 100%)', 'type': u'QUERY', 'waiting_time': u'52m8s'}, {'status': u'FINISHED', 'rows_fetched': 53, 'user': u'admin', 'canWrite': False, 'duration': 3369000.0, 'id': u'4d497267f34ff17d:817bdfb500000000', 'apiStatus': 'SUCCEEDED', 'name': u'select * from customers', 'submitted': u'2017-10-25 15:38:12.872825000', 'queue': u'root.admin', 'waiting': True, 'progress': u'2 / 3 (66.6667%)', 'type': u'QUERY', 'waiting_time': u'52m8s'}]
     for i in range(0,len(target)):
     for i in range(0,len(target)):
-      for key, value in target[i].iteritems():
+      for key, value in target[i].items():
         assert_equal(response.get('apps')[i].get(key), value)
         assert_equal(response.get('apps')[i].get(key), value)
 
 
   def test_app(self):
   def test_app(self):
     response = self.api.app('4d497267f34ff17d:817bdfb500000000')
     response = self.api.app('4d497267f34ff17d:817bdfb500000000')
     for key, value in {'status': u'FINISHED', 'name': u'select * from customers',
     for key, value in {'status': u'FINISHED', 'name': u'select * from customers',
       'duration': 3369000.0, 'progress': 66.6667, 'user': u'admin', 'type': 'queries',
       'duration': 3369000.0, 'progress': 66.6667, 'user': u'admin', 'type': 'queries',
-      'id': '4d497267f34ff17d:817bdfb500000000', 'submitted': u'2017-10-25 15:38:12.872825000', 'apiStatus': 'SUCCEEDED', 'doc_url': 'http://url.com/query_plan?query_id=4d497267f34ff17d:817bdfb500000000'}.iteritems():
+      'id': '4d497267f34ff17d:817bdfb500000000', 'submitted': u'2017-10-25 15:38:12.872825000', 'apiStatus': 'SUCCEEDED', 'doc_url': 'http://url.com/query_plan?query_id=4d497267f34ff17d:817bdfb500000000'}.items():
       assert_equal(response.get(key), value)
       assert_equal(response.get(key), value)
 
 
     response = self.api.app('8a46a8865624698f:b80b211500000000')
     response = self.api.app('8a46a8865624698f:b80b211500000000')
@@ -620,7 +622,7 @@ class TestImpalaApi(object):
     for key, value in {'status': u'FINISHED',
     for key, value in {'status': u'FINISHED',
       'name': u'SELECT sample_07.description, sample_07.salary FROM   sample...', 'duration': 3355000.0, 'progress': 100.0, 'user': u'admin',
       'name': u'SELECT sample_07.description, sample_07.salary FROM   sample...', 'duration': 3355000.0, 'progress': 100.0, 'user': u'admin',
       'type': 'queries', 'id': '8a46a8865624698f:b80b211500000000', 'submitted': u'2017-10-25 15:38:26.637010000',
       'type': 'queries', 'id': '8a46a8865624698f:b80b211500000000', 'submitted': u'2017-10-25 15:38:26.637010000',
-      'apiStatus': 'SUCCEEDED', 'doc_url': 'http://url.com/query_plan?query_id=8a46a8865624698f:b80b211500000000'}.iteritems():
+      'apiStatus': 'SUCCEEDED', 'doc_url': 'http://url.com/query_plan?query_id=8a46a8865624698f:b80b211500000000'}.items():
       assert_equal(response.get(key), value)
       assert_equal(response.get(key), value)
 
 
 
 
@@ -671,7 +673,7 @@ class TestSparkNoHadoop(object):
     assert_equal(response_log['logs']['logs'], 'dummy_logs')
     assert_equal(response_log['logs']['logs'], 'dummy_logs')
 
 
 
 
-class MockYarnApi:
+class MockYarnApi(object):
   def __init__(self, user, jt=None):
   def __init__(self, user, jt=None):
     self.user = user
     self.user = user
 
 
@@ -742,7 +744,7 @@ class HistoryServerHaApi(object):
   def __init__(self, username): pass
   def __init__(self, username): pass
 
 
 
 
-class MockResourceManagerApi:
+class MockResourceManagerApi(object):
   APPS = {
   APPS = {
     'application_1356251510842_0054': {
     'application_1356251510842_0054': {
         u'finishedTime': 1356961070119,
         u'finishedTime': 1356961070119,
@@ -866,7 +868,7 @@ class MockResourceManagerApi:
       u'app': MockResourceManagerApi.APPS[job_id]
       u'app': MockResourceManagerApi.APPS[job_id]
     }
     }
 
 
-class MockImpalaQueryApi:
+class MockImpalaQueryApi(object):
   APPS = {
   APPS = {
     '8a46a8865624698f:b80b211500000000': {u'stmt_type': u'QUERY', u'resource_pool': u'root.admin', u'waiting': True, u'last_event': u'Unregister query', u'start_time': u'2017-10-25 15:38:26.637010000', u'rows_fetched': 28, u'stmt': u'SELECT sample_07.description, sample_07.salary\r\nFROM\r\n  sample_07\r\nWHERE\r\n( sample_07.salary > 100000)\r\nORDER BY sample_07.salary DESC\r\nLIMIT 1000', u'executing': False, u'state': u'FINISHED', u'query_id': u'8a46a8865624698f:b80b211500000000', u'end_time': u'2017-10-25 16:34:22.592036000', u'duration': u'55m55s', u'progress': u'1 / 1 ( 100%)', u'effective_user': u'admin', u'default_db': u'default', u'waiting_time': u'52m8s'},
     '8a46a8865624698f:b80b211500000000': {u'stmt_type': u'QUERY', u'resource_pool': u'root.admin', u'waiting': True, u'last_event': u'Unregister query', u'start_time': u'2017-10-25 15:38:26.637010000', u'rows_fetched': 28, u'stmt': u'SELECT sample_07.description, sample_07.salary\r\nFROM\r\n  sample_07\r\nWHERE\r\n( sample_07.salary > 100000)\r\nORDER BY sample_07.salary DESC\r\nLIMIT 1000', u'executing': False, u'state': u'FINISHED', u'query_id': u'8a46a8865624698f:b80b211500000000', u'end_time': u'2017-10-25 16:34:22.592036000', u'duration': u'55m55s', u'progress': u'1 / 1 ( 100%)', u'effective_user': u'admin', u'default_db': u'default', u'waiting_time': u'52m8s'},
     '4d497267f34ff17d:817bdfb500000000': {u'stmt_type': u'QUERY', u'resource_pool': u'root.admin', u'waiting': True, u'last_event': u'Unregister query', u'start_time': u'2017-10-25 15:38:12.872825000', u'rows_fetched': 53, u'stmt': u'select * from customers', u'executing': False, u'state': u'FINISHED', u'query_id': u'4d497267f34ff17d:817bdfb500000000', u'end_time': u'2017-10-25 16:34:22.589811000', u'duration': u'56m9s', u'progress': u'2 / 3 (66.6667%)', u'effective_user': u'admin', u'default_db': u'default', u'waiting_time': u'52m8s'}
     '4d497267f34ff17d:817bdfb500000000': {u'stmt_type': u'QUERY', u'resource_pool': u'root.admin', u'waiting': True, u'last_event': u'Unregister query', u'start_time': u'2017-10-25 15:38:12.872825000', u'rows_fetched': 53, u'stmt': u'select * from customers', u'executing': False, u'state': u'FINISHED', u'query_id': u'4d497267f34ff17d:817bdfb500000000', u'end_time': u'2017-10-25 16:34:22.589811000', u'duration': u'56m9s', u'progress': u'2 / 3 (66.6667%)', u'effective_user': u'admin', u'default_db': u'default', u'waiting_time': u'52m8s'}

+ 33 - 28
apps/jobbrowser/src/jobbrowser/views.py

@@ -15,15 +15,20 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from past.builtins import cmp
+from future import standard_library
+standard_library.install_aliases()
+from builtins import filter
+from builtins import str
 import logging
 import logging
 import re
 import re
 import string
 import string
 import time
 import time
-import urllib2
-import urlparse
+import urllib.request, urllib.error, urllib.parse
+import urllib.parse
 
 
 from lxml import html
 from lxml import html
-from urllib import quote_plus
+from urllib.parse import quote_plus
 
 
 from django.http import HttpResponseRedirect
 from django.http import HttpResponseRedirect
 from django.utils.functional import wraps
 from django.utils.functional import wraps
@@ -72,7 +77,7 @@ def check_job_permission(view_func):
     jobid = kwargs['job']
     jobid = kwargs['job']
     try:
     try:
       job = get_job(request, job_id=jobid)
       job = get_job(request, job_id=jobid)
-    except ApplicationNotRunning, e:
+    except ApplicationNotRunning as e:
       LOG.warn('Job %s has not yet been accepted by the RM, will poll for status.' % jobid)
       LOG.warn('Job %s has not yet been accepted by the RM, will poll for status.' % jobid)
       return job_not_assigned(request, jobid, request.path)
       return job_not_assigned(request, jobid, request.path)
 
 
@@ -88,15 +93,15 @@ def check_job_permission(view_func):
 def get_job(request, job_id):
 def get_job(request, job_id):
   try:
   try:
     job = get_api(request.user, request.jt).get_job(jobid=job_id)
     job = get_api(request.user, request.jt).get_job(jobid=job_id)
-  except ApplicationNotRunning, e:
+  except ApplicationNotRunning as e:
     if e.job.get('state', '').lower() == 'accepted':
     if e.job.get('state', '').lower() == 'accepted':
       rm_api = resource_manager_api.get_resource_manager(request.user)
       rm_api = resource_manager_api.get_resource_manager(request.user)
       job = Application(e.job, rm_api)
       job = Application(e.job, rm_api)
     else:
     else:
       raise e  # Job has not yet been accepted by RM
       raise e  # Job has not yet been accepted by RM
-  except JobExpired, e:
+  except JobExpired as e:
     raise PopupException(_('Job %s has expired.') % job_id, detail=_('Cannot be found on the History Server.'))
     raise PopupException(_('Job %s has expired.') % job_id, detail=_('Cannot be found on the History Server.'))
-  except Exception, e:
+  except Exception as e:
     msg = 'Could not find job %s.'
     msg = 'Could not find job %s.'
     LOG.exception(msg % job_id)
     LOG.exception(msg % job_id)
     raise PopupException(_(msg) % job_id, detail=e)
     raise PopupException(_(msg) % job_id, detail=e)
@@ -118,9 +123,9 @@ def job_not_assigned(request, jobid, path):
     try:
     try:
       get_api(request.user, request.jt).get_job(jobid=jobid)
       get_api(request.user, request.jt).get_job(jobid=jobid)
       result['status'] = 0
       result['status'] = 0
-    except ApplicationNotRunning, e:
+    except ApplicationNotRunning as e:
       result['status'] = 1
       result['status'] = 1
-    except Exception, e:
+    except Exception as e:
       result['message'] = _('Error polling job %s: %s') % (jobid, e)
       result['message'] = _('Error polling job %s: %s') % (jobid, e)
 
 
     return JsonResponse(result, encoder=JSONEncoderForHTML)
     return JsonResponse(result, encoder=JSONEncoderForHTML)
@@ -149,7 +154,7 @@ def jobs(request):
           time_value=int(time_value),
           time_value=int(time_value),
           time_unit=time_unit
           time_unit=time_unit
       )
       )
-    except Exception, ex:
+    except Exception as ex:
       ex_message = str(ex)
       ex_message = str(ex)
       if 'Connection refused' in ex_message or 'standby RM' in ex_message:
       if 'Connection refused' in ex_message or 'standby RM' in ex_message:
         raise PopupException(_('Resource Manager cannot be contacted or might be down.'))
         raise PopupException(_('Resource Manager cannot be contacted or might be down.'))
@@ -287,7 +292,7 @@ def kill_job(request, job):
 
 
   try:
   try:
     job.kill()
     job.kill()
-  except Exception, e:
+  except Exception as e:
     LOG.exception('Killing job')
     LOG.exception('Killing job')
     raise PopupException(e)
     raise PopupException(e)
 
 
@@ -297,7 +302,7 @@ def kill_job(request, job):
   while time.time() - cur_time < 15:
   while time.time() - cur_time < 15:
     try:
     try:
       job = api.get_job(jobid=job.jobId)
       job = api.get_job(jobid=job.jobId)
-    except Exception, e:
+    except Exception as e:
       LOG.warn('Failed to get job with ID %s: %s' % (job.jobId, e))
       LOG.warn('Failed to get job with ID %s: %s' % (job.jobId, e))
     else:
     else:
       if job.status not in ["RUNNING", "QUEUED"]:
       if job.status not in ["RUNNING", "QUEUED"]:
@@ -320,7 +325,7 @@ def job_executor_logs(request, job, attempt_index=0, name='syslog', offset=LOG_O
       log = job.history_server_api.download_executors_logs(request, job, name, offset)
       log = job.history_server_api.download_executors_logs(request, job, name, offset)
     response['status'] = 0
     response['status'] = 0
     response['log'] = LinkJobLogs._make_hdfs_links(log)
     response['log'] = LinkJobLogs._make_hdfs_links(log)
-  except Exception, e:
+  except Exception as e:
     response['log'] = _('Failed to retrieve executor log: %s' % e)
     response['log'] = _('Failed to retrieve executor log: %s' % e)
 
 
   return JsonResponse(response)
   return JsonResponse(response)
@@ -360,9 +365,9 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=L
         log_link = app['amContainerLogs']
         log_link = app['amContainerLogs']
     elif app.get('amContainerLogs'):
     elif app.get('amContainerLogs'):
       log_link = app.get('amContainerLogs')
       log_link = app.get('amContainerLogs')
-  except (KeyError, RestException), e:
+  except (KeyError, RestException) as e:
     raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
     raise KeyError(_("Cannot find job attempt '%(id)s'.") % {'id': job.jobId}, e)
-  except Exception, e:
+  except Exception as e:
     raise Exception(_("Failed to get application for job %s: %s") % (job.jobId, e))
     raise Exception(_("Failed to get application for job %s: %s") % (job.jobId, e))
 
 
   if log_link:
   if log_link:
@@ -374,7 +379,7 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=L
     if offset != 0:
     if offset != 0:
       params['start'] = offset
       params['start'] = offset
 
 
-    root = Resource(get_log_client(log_link), urlparse.urlsplit(log_link)[2], urlencode=False)
+    root = Resource(get_log_client(log_link), urllib.parse.urlsplit(log_link)[2], urlencode=False)
     api_resp = None
     api_resp = None
 
 
     try:
     try:
@@ -383,7 +388,7 @@ def job_attempt_logs_json(request, job, attempt_index=0, name='syslog', offset=L
 
 
       response['status'] = 0
       response['status'] = 0
       response['log'] = LinkJobLogs._make_hdfs_links(log, is_embeddable)
       response['log'] = LinkJobLogs._make_hdfs_links(log, is_embeddable)
-    except Exception, e:
+    except Exception as e:
       response['log'] = _('Failed to retrieve log: %s' % e)
       response['log'] = _('Failed to retrieve log: %s' % e)
       try:
       try:
         debug_info = '\nLog Link: %s' % log_link
         debug_info = '\nLog Link: %s' % log_link
@@ -501,7 +506,7 @@ def single_task_attempt(request, job, taskid, attemptid):
 
 
   try:
   try:
     attempt = task.get_attempt(attemptid)
     attempt = task.get_attempt(attemptid)
-  except (KeyError, RestException), e:
+  except (KeyError, RestException) as e:
     raise PopupException(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
     raise PopupException(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
 
 
   return render("attempt.mako", request, {
   return render("attempt.mako", request, {
@@ -520,7 +525,7 @@ def single_task_attempt_logs(request, job, taskid, attemptid, offset=LOG_OFFSET_
 
 
   try:
   try:
     attempt = task.get_attempt(attemptid)
     attempt = task.get_attempt(attemptid)
-  except (KeyError, RestException), e:
+  except (KeyError, RestException) as e:
     raise KeyError(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
     raise KeyError(_("Cannot find attempt '%(id)s' in task") % {'id': attemptid}, e)
 
 
   first_log_tab = 0
   first_log_tab = 0
@@ -539,7 +544,7 @@ def single_task_attempt_logs(request, job, taskid, attemptid, offset=LOG_OFFSET_
     log_tab = [i for i, log in enumerate(logs) if log]
     log_tab = [i for i, log in enumerate(logs) if log]
     if log_tab:
     if log_tab:
       first_log_tab = log_tab[0]
       first_log_tab = log_tab[0]
-  except urllib2.URLError:
+  except urllib.error.URLError:
     logs = [_("Failed to retrieve log. TaskTracker not ready.")] * 4
     logs = [_("Failed to retrieve log. TaskTracker not ready.")] * 4
 
 
   context = {
   context = {
@@ -603,7 +608,7 @@ def single_tracker(request, trackerid):
 
 
   try:
   try:
     tracker = jt.get_tracker(trackerid)
     tracker = jt.get_tracker(trackerid)
-  except Exception, e:
+  except Exception as e:
     raise PopupException(_('The tracker could not be contacted.'), detail=e)
     raise PopupException(_('The tracker could not be contacted.'), detail=e)
   return render("tasktracker.mako", request, {'tracker':tracker})
   return render("tasktracker.mako", request, {'tracker':tracker})
 
 
@@ -612,7 +617,7 @@ def container(request, node_manager_http_address, containerid):
 
 
   try:
   try:
     tracker = jt.get_tracker(node_manager_http_address, containerid)
     tracker = jt.get_tracker(node_manager_http_address, containerid)
-  except Exception, e:
+  except Exception as e:
     # TODO: add a redirect of some kind
     # TODO: add a redirect of some kind
     raise PopupException(_('The container disappears as soon as the job finishes.'), detail=e)
     raise PopupException(_('The container disappears as soon as the job finishes.'), detail=e)
   return render("container.mako", request, {'tracker':tracker})
   return render("container.mako", request, {'tracker':tracker})
@@ -663,7 +668,7 @@ def make_substitutions(conf):
         s = s.replace(substr, sub(conf[g], depth+1))
         s = s.replace(substr, sub(conf[g], depth+1))
     return s
     return s
 
 
-  for k, v in conf.items():
+  for k, v in list(conf.items()):
     conf[k] = sub(v)
     conf[k] = sub(v)
   return conf
   return conf
 
 
@@ -707,7 +712,7 @@ def get_state_link(request, option=None, val='', VALID_OPTIONS = ("state", "user
   if option is not None:
   if option is not None:
     states[option] = val
     states[option] = val
 
 
-  return "&".join([ "%s=%s" % (key, quote_plus(value)) for key, value in states.iteritems() ])
+  return "&".join([ "%s=%s" % (key, quote_plus(value)) for key, value in states.items() ])
 
 
 
 
 ## All Unused below
 ## All Unused below
@@ -767,10 +772,10 @@ def jobbrowser(request):
 
 
   status = request.jt.cluster_status()
   status = request.jt.cluster_status()
   alljobs = [] #get_matching_jobs(request)
   alljobs = [] #get_matching_jobs(request)
-  runningjobs = filter(check_job_state('RUNNING'), alljobs)
-  completedjobs = filter(check_job_state('COMPLETED'), alljobs)
-  failedjobs = filter(check_job_state('FAILED'), alljobs)
-  killedjobs = filter(check_job_state('KILLED'), alljobs)
+  runningjobs = list(filter(check_job_state('RUNNING'), alljobs))
+  completedjobs = list(filter(check_job_state('COMPLETED'), alljobs))
+  failedjobs = list(filter(check_job_state('FAILED'), alljobs))
+  killedjobs = list(filter(check_job_state('KILLED'), alljobs))
   jobqueues = request.jt.queues()
   jobqueues = request.jt.queues()
 
 
   return render("jobbrowser.html", request, {
   return render("jobbrowser.html", request, {

+ 32 - 26
apps/jobbrowser/src/jobbrowser/yarn_models.py

@@ -15,11 +15,17 @@
 # See the License for the specific language governing permissions and
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # limitations under the License.
 
 
+from __future__ import division
+from future import standard_library
+standard_library.install_aliases()
+from builtins import str
+from past.utils import old_div
+from builtins import object
 import logging
 import logging
 import os
 import os
 import re
 import re
 import time
 import time
-import urlparse
+import urllib.parse
 
 
 from lxml import html
 from lxml import html
 
 
@@ -33,7 +39,7 @@ from desktop.lib.view_util import big_filesizeformat, format_duration_in_millis
 from hadoop import cluster
 from hadoop import cluster
 from hadoop.yarn.clients import get_log_client
 from hadoop.yarn.clients import get_log_client
 
 
-from itertools import izip
+
 
 
 from jobbrowser.models import format_unixtime_ms
 from jobbrowser.models import format_unixtime_ms
 
 
@@ -45,7 +51,7 @@ class Application(object):
 
 
   def __init__(self, attrs, rm_api=None):
   def __init__(self, attrs, rm_api=None):
     self.api = rm_api
     self.api = rm_api
-    for attr in attrs.keys():
+    for attr in list(attrs.keys()):
       setattr(self, attr, attrs[attr])
       setattr(self, attr, attrs[attr])
 
 
     self._fixup()
     self._fixup()
@@ -145,7 +151,7 @@ class SparkJob(Application):
         actual_url = actual_url.strip('/').replace('jobs', '')
         actual_url = actual_url.strip('/').replace('jobs', '')
       self.trackingUrl = actual_url
       self.trackingUrl = actual_url
       LOG.debug("SparkJob tracking URL: %s" % self.trackingUrl)
       LOG.debug("SparkJob tracking URL: %s" % self.trackingUrl)
-    except Exception, e:
+    except Exception as e:
       LOG.warn("Failed to resolve Spark Job's actual tracking URL: %s" % e)
       LOG.warn("Failed to resolve Spark Job's actual tracking URL: %s" % e)
     finally:
     finally:
       if resp is not None:
       if resp is not None:
@@ -155,7 +161,7 @@ class SparkJob(Application):
     response = None
     response = None
     try:
     try:
       response = function(*args, **kwargs)
       response = function(*args, **kwargs)
-    except Exception, e:
+    except Exception as e:
       LOG.warn('Spark resolve tracking URL returned a failed response: %s' % e)
       LOG.warn('Spark resolve tracking URL returned a failed response: %s' % e)
     return response
     return response
 
 
@@ -195,7 +201,7 @@ class SparkJob(Application):
             big_filesizeformat(e.get('totalShuffleWrite', 0)),
             big_filesizeformat(e.get('totalShuffleWrite', 0)),
             e.get('executorLogs', '')
             e.get('executorLogs', '')
           ])
           ])
-    except Exception, e:
+    except Exception as e:
       LOG.error('Failed to get Spark Job executors: %s' % e)
       LOG.error('Failed to get Spark Job executors: %s' % e)
       # Prevent a nosedive. Don't create metrics if api changes or url is unreachable.
       # Prevent a nosedive. Don't create metrics if api changes or url is unreachable.
 
 
@@ -206,7 +212,7 @@ class SparkJob(Application):
       headers = ['executor_id', 'address', 'rdd_blocks', 'storage_memory', 'disk_used', 'active_tasks', 'failed_tasks',
       headers = ['executor_id', 'address', 'rdd_blocks', 'storage_memory', 'disk_used', 'active_tasks', 'failed_tasks',
                  'complete_tasks', 'task_time', 'input', 'shuffle_read', 'shuffle_write', 'logs']
                  'complete_tasks', 'task_time', 'input', 'shuffle_read', 'shuffle_write', 'logs']
       for executor in executors:
       for executor in executors:
-        executor_data = dict(izip(headers, executor))
+        executor_data = dict(zip(headers, executor))
         executor_data.update({'id': executor_data['executor_id'] + '_executor_' + self.jobId, 'type': 'SPARK_EXECUTOR'})
         executor_data.update({'id': executor_data['executor_id'] + '_executor_' + self.jobId, 'type': 'SPARK_EXECUTOR'})
         executor_list.append(executor_data)
         executor_list.append(executor_data)
     return executor_list
     return executor_list
@@ -217,7 +223,7 @@ class Job(object):
   def __init__(self, api, attrs):
   def __init__(self, api, attrs):
     self.api = api
     self.api = api
     self.is_mr2 = True
     self.is_mr2 = True
-    for attr in attrs.keys():
+    for attr in list(attrs.keys()):
       if attr == 'acls':
       if attr == 'acls':
         # 'acls' are actually not available in the API
         # 'acls' are actually not available in the API
         LOG.warn('Not using attribute: %s' % attrs[attr])
         LOG.warn('Not using attribute: %s' % attrs[attr])
@@ -246,7 +252,7 @@ class Job(object):
 
 
       if self.desiredReduces > 0:
       if self.desiredReduces > 0:
         if self.progress is not None:
         if self.progress is not None:
-          self.progress = int((self.progress + self.reduces_percent_complete) / 2)
+          self.progress = int(old_div((self.progress + self.reduces_percent_complete), 2))
         else:
         else:
           self.progress = self.reduces_percent_complete
           self.progress = self.reduces_percent_complete
 
 
@@ -304,7 +310,7 @@ class Job(object):
       try:
       try:
         conf = self.api.conf(self.id)
         conf = self.api.conf(self.id)
         self._full_job_conf = conf['conf']
         self._full_job_conf = conf['conf']
-      except TypeError, e:
+      except TypeError as e:
         LOG.exception('YARN API call failed to return all the data: %s' % conf)
         LOG.exception('YARN API call failed to return all the data: %s' % conf)
     return self._full_job_conf
     return self._full_job_conf
 
 
@@ -312,7 +318,7 @@ class Job(object):
   def conf_keys(self):
   def conf_keys(self):
     try:
     try:
       return dict([(line['name'], line['value']) for line in self.full_job_conf['property']])
       return dict([(line['name'], line['value']) for line in self.full_job_conf['property']])
-    except Exception, e:
+    except Exception as e:
       LOG.error('Failed to parse conf_keys from YARN job configuration.')
       LOG.error('Failed to parse conf_keys from YARN job configuration.')
       return None
       return None
 
 
@@ -339,7 +345,7 @@ class Job(object):
 class YarnV2Job(Job):
 class YarnV2Job(Job):
   def __init__(self, api, attrs):
   def __init__(self, api, attrs):
     self.api = api
     self.api = api
-    for attr in attrs.keys():
+    for attr in list(attrs.keys()):
       if attr == 'acls':
       if attr == 'acls':
         # 'acls' are actually not available in the API
         # 'acls' are actually not available in the API
         LOG.warn('Not using attribute: %s' % attrs[attr])
         LOG.warn('Not using attribute: %s' % attrs[attr])
@@ -388,7 +394,7 @@ class YarnV2Job(Job):
     setattr(self, 'finishTime', finishTime)
     setattr(self, 'finishTime', finishTime)
 
 
     try:
     try:
-      setattr(self, 'assignedContainerId', urlparse.urlsplit(self.amContainerLogs).path.split('/node/containerlogs/')[1].split('/')[0])
+      setattr(self, 'assignedContainerId', urllib.parse.urlsplit(self.amContainerLogs).path.split('/node/containerlogs/')[1].split('/')[0])
     except Exception:
     except Exception:
       setattr(self, 'assignedContainerId', '')
       setattr(self, 'assignedContainerId', '')
 
 
@@ -412,7 +418,7 @@ class YarnV2Job(Job):
     return self._job_attempts
     return self._job_attempts
 
 
 # There's are tasks for Oozie workflow so we create a dummy one.
 # There's are tasks for Oozie workflow so we create a dummy one.
-class YarnTask:
+class YarnTask(object):
   def __init__(self, job):
   def __init__(self, job):
     self.job = job
     self.job = job
 
 
@@ -458,12 +464,12 @@ class KilledJob(Job):
     return {'jobAttempt': []}
     return {'jobAttempt': []}
 
 
 
 
-class Task:
+class Task(object):
 
 
   def __init__(self, job, attrs):
   def __init__(self, job, attrs):
     self.job = job
     self.job = job
     if attrs:
     if attrs:
-      for key, value in attrs.iteritems():
+      for key, value in attrs.items():
         setattr(self, key, value)
         setattr(self, key, value)
     self.is_mr2 = True
     self.is_mr2 = True
 
 
@@ -509,12 +515,12 @@ class Task:
     return Attempt(self, json)
     return Attempt(self, json)
 
 
 
 
-class Attempt:
+class Attempt(object):
 
 
   def __init__(self, task, attrs):
   def __init__(self, task, attrs):
     self.task = task
     self.task = task
     if attrs:
     if attrs:
-      for key, value in attrs.iteritems():
+      for key, value in attrs.items():
         setattr(self, key, value)
         setattr(self, key, value)
     self.is_mr2 = True
     self.is_mr2 = True
     self._fixup()
     self._fixup()
@@ -607,11 +613,11 @@ class Attempt:
       'doAs': user
       'doAs': user
     }
     }
     log_link = re.sub('job_[^/]+', str(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)
+    root = Resource(get_log_client(log_link), urllib.parse.urlsplit(log_link)[2], urlencode=False)
     response = root.get('/', params=params)
     response = root.get('/', params=params)
     links = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]//a/@href')
     links = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]//a/@href')
-    parsed_links = map(lambda x: urlparse.urlsplit(x), links)
-    return map(lambda x: x and len(x) >= 2 and x[2].split('/')[-2] or '', parsed_links)
+    parsed_links = [urllib.parse.urlsplit(x) for x in links]
+    return [x and len(x) >= 2 and x[2].split('/')[-2] or '' for x in parsed_links]
 
 
   def get_task_log(self, offset=0):
   def get_task_log(self, offset=0):
     logs = []
     logs = []
@@ -637,10 +643,10 @@ class Attempt:
       response = None
       response = None
       try:
       try:
         log_link = re.sub('job_[^/]+', str(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)
+        root = Resource(get_log_client(log_link), urllib.parse.urlsplit(log_link)[2], urlencode=False)
         response = root.get(link, params=params)
         response = root.get(link, params=params)
         log = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
         log = html.fromstring(response, parser=html.HTMLParser()).xpath('/html/body/table/tbody/tr/td[2]')[0].text_content()
-      except Exception, e:
+      except Exception as e:
         log = _('Failed to retrieve log: %s' % e)
         log = _('Failed to retrieve log: %s' % e)
         try:
         try:
           debug_info = '\nLog Link: %s' % log_link
           debug_info = '\nLog Link: %s' % log_link
@@ -658,7 +664,7 @@ class YarnV2Attempt(Attempt):
   def __init__(self, task, attrs):
   def __init__(self, task, attrs):
     self.task = task
     self.task = task
     if attrs:
     if attrs:
-      for key, value in attrs.iteritems():
+      for key, value in attrs.items():
         setattr(self, key, value)
         setattr(self, key, value)
     self.is_mr2 = True
     self.is_mr2 = True
     self._fixup()
     self._fixup()
@@ -684,11 +690,11 @@ class YarnV2Attempt(Attempt):
     setattr(self, 'status', 'RUNNING' if self.finishedTime == 0 else 'SUCCEEDED')
     setattr(self, 'status', 'RUNNING' if self.finishedTime == 0 else 'SUCCEEDED')
     setattr(self, 'properties', {})
     setattr(self, 'properties', {})
 
 
-class Container:
+class Container(object):
 
 
   def __init__(self, attrs):
   def __init__(self, attrs):
     if attrs:
     if attrs:
-      for key, value in attrs['container'].iteritems():
+      for key, value in attrs['container'].items():
         setattr(self, key, value)
         setattr(self, key, value)
     self.is_mr2 = True
     self.is_mr2 = True