<commit_before>#!/usr/bin/env python
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import logging

from django.utils.translation import ugettext as _
from hadoop.yarn import resource_manager_api

from desktop.lib.exceptions import MessageException
from desktop.lib.exceptions_renderable import PopupException
from jobbrowser.conf import MAX_JOB_FETCH
from jobbrowser.views import job_single_logs


LOG = logging.getLogger(__name__)


try:
  from jobbrowser.api import YarnApi as NativeYarnApi, ApplicationNotRunning, JobExpired
  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.yarn_models import Application
except Exception, e:
  LOG.exception('Some application are not enabled: %s' % e)


class JobApi(Api):

  def __init__(self, user):
    self.user = user
    self.yarn_api = YarnApi(user) # TODO: actually long term move job aggregations to the frontend instead probably
    self.impala_api = ImpalaApi(user)
    self.request = None

  def apps(self, filters):
    jobs = self.yarn_api.apps(filters)
    # += Impala
    # += Sqoop2
    return jobs

  def app(self, appid):
    return self._get_api(appid).app(appid)

  def action(self, app_ids, operation):
    return self._get_api(app_ids).action(operation, app_ids)

  def logs(self, appid, app_type, log_name):
    return self._get_api(appid).logs(appid, app_type, log_name)

  def profile(self, appid, app_type, app_property, app_filters):
    return self._get_api(appid).profile(appid, app_type, app_property, app_filters)

  def _get_api(self, appid):
    if type(appid) == list:
      return self.yarn_api
    elif appid.startswith('task_'):
      return YarnMapReduceTaskApi(self.user, appid)
    elif appid.startswith('attempt_'):
      return YarnMapReduceTaskAttemptApi(self.user, appid)
    else:
      return self.yarn_api # application_

  def _set_request(self, request):
    self.request = request


class YarnApi(Api):
  """YARN, MR, Spark"""

  def apps(self, filters):
    filter_params = {
      'user': self.user,
      'username': '',
      'text': '',
      'state': 'all',
      'states': ''
    }

    filter_params.update(_extract_query_params(filters))

    if filters.get('states'):
      filter_params['states'] = filters['states']

    if 'time' in filters:
      filter_params['time_value'] = int(filters['time']['time_value'])
      filter_params['time_unit'] = filters['time']['time_unit']

    filter_params['limit'] = MAX_JOB_FETCH.get()

    jobs = NativeYarnApi(self.user).get_jobs(**filter_params)

    apps = [massage_job_for_json(job, user=self.user) for job in sorted(jobs, key=lambda job: job.jobId, reverse=True)]

    return {
      'apps': [{
        'id': app['id'],
        'name': app['name'],
        'type': app['applicationType'],
        'status': app['status'],
        'apiStatus': self._api_status(app['status']),
        'user': app['user'],
        'progress': app['progress'],
        'queue': app['queueName'],
        'duration': app['durationMs'],
        'submitted': app['startTimeMs'],
        'canWrite': app['canKill'],
      } for app in apps],
      'total': len(apps)
    }


  def app(self, appid):
    try:
      job = NativeYarnApi(self.user).get_job(jobid=appid)
    except ApplicationNotRunning, e:
      if e.job.get('state', '').lower() == 'accepted':
        rm_api = resource_manager_api.get_resource_manager(self.user)
        job = Application(e.job, rm_api)
      else:
        raise e  # Job has not yet been accepted by RM
    except JobExpired, e:
      raise PopupException(_('Job %s has expired.') % appid, detail=_('Cannot be found on the History Server.'))
    except Exception, e:
      msg = 'Could not find job %s.'
      LOG.exception(msg % appid)
      raise PopupException(_(msg) % appid, detail=e)


    app = massage_job_for_json(job, user=self.user)

    common = {
        'id': app['id'],
        'name': app['name'],
        'type': app['applicationType'],
        'status': app['status'],
        'apiStatus': self._api_status(app['status']),
        'user': app['user'],
        'progress': app['progress'],
        'duration': app['durationMs'],
        'submitted': app['startTimeMs'],
        'canWrite': app['canKill'],
    }

    if app['applicationType'] == 'MR2' or app['applicationType'] == 'MAPREDUCE':
      common['type'] = 'MAPREDUCE'

      common['properties'] = {
          'maps_percent_complete': app['mapsPercentComplete'] or 100,
          'reduces_percent_complete': app['reducesPercentComplete'] or 100,
          'finishedMaps': app['finishedMaps'],
          'finishedReduces': app['finishedReduces'],
          'desiredMaps': app['desiredMaps'],
          'desiredReduces': app['desiredReduces'],
          'durationFormatted': app['durationFormatted'],
          'startTimeFormatted': app['startTimeFormatted'],

          'tasks': [],
          'metadata': [],
          'counters': []
      }
    elif app['applicationType'] == 'SPARK':
      common['type'] = 'SPARK'
      common['properties'] = {
        'metadata': [{'name': name, 'value': value} for name, value in app.iteritems()]<commit_msg>HUE-6727 [jb] 0/1 map progress percent gets listed as 100%<commit_after>#!/usr/bin/env python
# Licensed to Cloudera, Inc. under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  Cloudera, Inc. licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import logging

from django.utils.translation import ugettext as _
from hadoop.yarn import resource_manager_api

from desktop.lib.exceptions import MessageException
from desktop.lib.exceptions_renderable import PopupException
from jobbrowser.conf import MAX_JOB_FETCH
from jobbrowser.views import job_single_logs


LOG = logging.getLogger(__name__)


try:
  from jobbrowser.api import YarnApi as NativeYarnApi, ApplicationNotRunning, JobExpired
  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.yarn_models import Application
except Exception, e:
  LOG.exception('Some application are not enabled: %s' % e)


class JobApi(Api):

  def __init__(self, user):
    self.user = user
    self.yarn_api = YarnApi(user) # TODO: actually long term move job aggregations to the frontend instead probably
    self.impala_api = ImpalaApi(user)
    self.request = None

  def apps(self, filters):
    jobs = self.yarn_api.apps(filters)
    # += Impala
    # += Sqoop2
    return jobs

  def app(self, appid):
    return self._get_api(appid).app(appid)

  def action(self, app_ids, operation):
    return self._get_api(app_ids).action(operation, app_ids)

  def logs(self, appid, app_type, log_name):
    return self._get_api(appid).logs(appid, app_type, log_name)

  def profile(self, appid, app_type, app_property, app_filters):
    return self._get_api(appid).profile(appid, app_type, app_property, app_filters)

  def _get_api(self, appid):
    if type(appid) == list:
      return self.yarn_api
    elif appid.startswith('task_'):
      return YarnMapReduceTaskApi(self.user, appid)
    elif appid.startswith('attempt_'):
      return YarnMapReduceTaskAttemptApi(self.user, appid)
    else:
      return self.yarn_api # application_

  def _set_request(self, request):
    self.request = request


class YarnApi(Api):
  """YARN, MR, Spark"""

  def apps(self, filters):
    filter_params = {
      'user': self.user,
      'username': '',
      'text': '',
      'state': 'all',
      'states': ''
    }

    filter_params.update(_extract_query_params(filters))

    if filters.get('states'):
      filter_params['states'] = filters['states']

    if 'time' in filters:
      filter_params['time_value'] = int(filters['time']['time_value'])
      filter_params['time_unit'] = filters['time']['time_unit']

    filter_params['limit'] = MAX_JOB_FETCH.get()

    jobs = NativeYarnApi(self.user).get_jobs(**filter_params)

    apps = [massage_job_for_json(job, user=self.user) for job in sorted(jobs, key=lambda job: job.jobId, reverse=True)]

    return {
      'apps': [{
        'id': app['id'],
        'name': app['name'],
        'type': app['applicationType'],
        'status': app['status'],
        'apiStatus': self._api_status(app['status']),
        'user': app['user'],
        'progress': app['progress'],
        'queue': app['queueName'],
        'duration': app['durationMs'],
        'submitted': app['startTimeMs'],
        'canWrite': app['canKill'],
      } for app in apps],
      'total': len(apps)
    }


  def app(self, appid):
    try:
      job = NativeYarnApi(self.user).get_job(jobid=appid)
    except ApplicationNotRunning, e:
      if e.job.get('state', '').lower() == 'accepted':
        rm_api = resource_manager_api.get_resource_manager(self.user)
        job = Application(e.job, rm_api)
      else:
        raise e  # Job has not yet been accepted by RM
    except JobExpired, e:
      raise PopupException(_('Job %s has expired.') % appid, detail=_('Cannot be found on the History Server.'))
    except Exception, e:
      msg = 'Could not find job %s.'
      LOG.exception(msg % appid)
      raise PopupException(_(msg) % appid, detail=e)


    app = massage_job_for_json(job, user=self.user)

    common = {
        'id': app['id'],
        'name': app['name'],
        'type': app['applicationType'],
        'status': app['status'],
        'apiStatus': self._api_status(app['status']),
        'user': app['user'],
        'progress': app['progress'],
        'duration': app['durationMs'],
        'submitted': app['startTimeMs'],
        'canWrite': app['canKill'],
    }

    if app['applicationType'] == 'MR2' or app['applicationType'] == 'MAPREDUCE':
      common['type'] = 'MAPREDUCE'

      if app['desiredMaps'] is None or app['finishedMaps'] is None:
        app['mapsPercentComplete'] = 100
      if app['desiredReduces'] is None or app['finishedReduces'] is None:
        app['reducesPercentComplete'] = 100

      common['properties'] = {
          'maps_percent_complete': app['mapsPercentComplete'] or 0,
          'reduces_percent_complete': app['reducesPercentComplete'] or 0,
          'finishedMaps': app['finishedMaps'] or 0,
          'finishedReduces': app['finishedReduces'] or 0,
          'desiredMaps': app['desiredMaps'] or 0,
          'desiredReduces': app['desiredReduces'] or 0,
          'durationFormatted': app['durationFormatted'],
          'startTimeFormatted': app['startTimeFormatted'],

          'tasks': [],
          'metadata': [],
          'counters': []
      }
    elif app['applicationType'] == 'SPARK':
      common['type'] = 'SPARK'
      common['properties'] = {
        'metadata': [{'name': name, 'value': value} for name, value in app.iteritems()]
{'fraction_non_alphanumeric': 0.1297221994591494, 'fraction_numerical': 0.003195935425714988, 'mean_word_length': 3.504983388704319, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5992326', 'n_tokens_mistral': 3814, 'n_tokens_neox': 3721, 'n_words': 988}