<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 itertools
import logging
import re
import time
from datetime import datetime

from django.utils.translation import ugettext as _

from jobbrowser.apis.base_api import Api
from libanalyze import analyze as analyzer, rules

ANALYZER = rules.TopDownAnalysis() # We need to parse some files so save as global
LOG = logging.getLogger(__name__)

try:
  from beeswax.models import Session
  from impala.server import get_api as get_impalad_api, _get_impala_server_url
  from impala.dbms import _get_server_name
except Exception, e:
  LOG.exception('Some application are not enabled: %s' % e)


def _get_api(user, cluster=None):
  if cluster and cluster.get('type') == 'altus-dw':
    server_url = 'http://impala-coordinator-%(name)s:25000' % cluster
  else:
    session = Session.objects.get_session(user, application=_get_server_name(cluster))
    server_url = _get_impala_server_url(session)
  return get_impalad_api(user=user, url=server_url)


class QueryApi(Api):

  def __init__(self, user, impala_api=None, cluster=None):
    if impala_api:
      self.api = impala_api
    else:
      self.api = _get_api(user, cluster)

  def apps(self, filters):
    kwargs = {}

    jobs = self.api.get_queries(**kwargs)

    filter_list = self._get_filter_list(filters)
    jobs_iter = itertools.chain(jobs['in_flight_queries'], jobs['completed_queries'])
    jobs_iter_filtered = self._n_filter(filter_list, jobs_iter)

    apps = {
      'apps': sorted([{
        'id': job['query_id'],
        'name': job['stmt'].replace('\r\n', ' ')[:60] + ('...' if len(job['stmt']) > 60 else ''),
        'status': job['state'],
        'apiStatus': self._api_status(job['state']),
        'type': job['stmt_type'],
        'user': job['effective_user'],
        'queue': job.get('resource_pool'),
        'progress': job['progress'],
        'isRunning': job['start_time'] > job['end_time'],
        'canWrite': job in jobs['in_flight_queries'],
        'duration': self._time_in_ms_groups(re.search(r"\s*(([\d.]*)([a-z]*))(([\d.]*)([a-z]*))?(([\d.]*)([a-z]*))?", job['duration'], re.MULTILINE).groups()),
        'submitted': job['start_time'],
        # Extra specific
        'rows_fetched': job['rows_fetched'],
        'waiting': job['waiting'],
        'waiting_time': job['waiting_time']
      } for job in jobs_iter_filtered], key=lambda job: job.get('submitted'), reverse=True),
      'total': 0
    }
    apps['total'] = len(apps['apps'])

    return apps

  def _time_in_ms_groups(self, groups):
    time = 0
    for x in range(0, len(groups), 3):
      if groups[x+1]:
        time += self._time_in_ms(groups[x+1], groups[x+2])
    return time

  def _time_in_ms(self, time, period):
    if period == 'ns':
      return float(time) / 1000
    elif period == 'ms':
      return float(time)
    elif period == 's':
      return float(time) * 1000
    elif period == 'm':
      return float(time) * 60000 #1000*60
    elif period == 'h':
      return float(time) * 3600000 #1000*60*60
    elif period == 'd':
      return float(time) * 86400000  # 1000*60*60*24
    else:
      return float(time)

  def app(self, appid):
    apps = self.apps({
      'text': 'id:' + appid
    })

    if not apps.get('apps'):
      return {
        'name': _('Unknown or expired query id %s') % appid
      }
    app = apps.get('apps')[0]
    progress_groups = re.search(r"([\d\.\,]+)%", app.get('progress'))
    app.update({
      'progress': float(progress_groups.group(1)) if progress_groups and progress_groups.group(1) else 100 if self._api_status(app.get('status')) in ['SUCCEEDED', 'FAILED'] else 1,
      'type': 'queries',
      'doc_url': "%s/query_plan?query_id=%s" % (self.api.url, appid),
      'properties': {
        'memory': '',
        'profile': '',
        'plan': '',
        'backends': '',
        'finstances': '',
        'metrics': ''
      }
    })

    return app

  def action(self, appid, action):
    message = {'message': '', 'status': 0}

    if action.get('action') == 'kill':

      for _id in appid:
        result = self.api.kill(_id)
        if result.get('error'):
          message['message'] = result.get('error')
          message['status'] = -1
        elif result.get('contents') and message.get('status') != -1:
          message['message'] = result.get('contents')

    return message;


  def logs(self, appid, app_type, log_name=None, is_embeddable=False):
    return {'logs': ''}

  def profile(self, appid, app_type, app_property, app_filters):
    if app_property == 'memory':
      return self._memory(appid, app_type, app_property, app_filters)
    elif app_property == 'profile':
      return self._query_profile(appid)
    elif app_property == 'backends':
      return self._query_backends(appid)
    elif app_property == 'finstances':
      return self._query_finstances(appid)
    else:
      return self._query(appid)

  def profile_encoded(self, appid):
    return self.api.get_query_profile_encoded(query_id=appid)

  def _memory(self, appid, app_type, app_property, app_filters):
    return self.api.get_query_memory(query_id=appid);

  def _metrics(self, appid):
    query_profile = self.api.get_query_profile_encoded(appid)
    profile = analyzer.analyze(analyzer.parse_data(query_profile))
    ANALYZER.pre_process(profile)
    metrics = analyzer.metrics(profile)

    result = ANALYZER.run(profile)
    if result and result[0]:
      for factor in result[0]['result']:
        if factor['reason'] and factor['result_id'] and metrics['nodes'].get(factor['result_id']):
          metrics['nodes'][factor['result_id']]['health'] = factor['reason']
    return metrics

  def _query(self, appid):
    query = self.api.get_query(query_id=appid)
    query['summary'] = query.get('summary').strip() if query.get('summary') else ''
    query['plan'] = query.get('plan').strip() if query.get('plan') else ''
    try:
      query['metrics'] = self._metrics(appid)
    except Exception, e:
      query['metrics'] = {'nodes' : {}}
      LOG.exception('Could not parse profile: %s' % e)

    if query.get('plan_json'):
      def get_exchange_icon (o):
        if re.search(r'broadcast', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-broadcast' }
        elif re.search(r'hash', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-random' }
        else:
          return { 'svg': 'hi-exchange' }
      def get_sigma_icon (o):
        if re.search(r'streaming', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-sigma' }
        else:
          return { 'svg': 'hi-sigma' }
      mapping = {
        'TOP-N': { 'type': 'TOPN', 'icon': { 'svg': 'hi-filter' } },
        'SORT': { 'type': 'SORT', 'icon': { 'svg': 'hi-sort' } },
        'MERGING-EXCHANGE': {'type': 'EXCHANGE', 'icon': { 'fn': get_exchange_icon } },
        'EXCHANGE': { 'type': 'EXCHANGE', 'icon': { 'fn': get_exchange_icon } },
        'SCAN HDFS': { 'type': 'SCAN_HDFS', 'icon': { 'svg': 'hi-copy' } },<commit_msg>HUE-8704 [jb] Hide profile health risks behind flag.<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 itertools
import logging
import re
import time
from datetime import datetime

from django.utils.translation import ugettext as _

from jobbrowser.apis.base_api import Api
from libanalyze import analyze as analyzer, rules
from notebook.conf import ENABLE_QUERY_ANALYSIS

ANALYZER = rules.TopDownAnalysis() # We need to parse some files so save as global
LOG = logging.getLogger(__name__)

try:
  from beeswax.models import Session
  from impala.server import get_api as get_impalad_api, _get_impala_server_url
  from impala.dbms import _get_server_name
except Exception, e:
  LOG.exception('Some application are not enabled: %s' % e)


def _get_api(user, cluster=None):
  if cluster and cluster.get('type') == 'altus-dw':
    server_url = 'http://impala-coordinator-%(name)s:25000' % cluster
  else:
    session = Session.objects.get_session(user, application=_get_server_name(cluster))
    server_url = _get_impala_server_url(session)
  return get_impalad_api(user=user, url=server_url)


class QueryApi(Api):

  def __init__(self, user, impala_api=None, cluster=None):
    if impala_api:
      self.api = impala_api
    else:
      self.api = _get_api(user, cluster)

  def apps(self, filters):
    kwargs = {}

    jobs = self.api.get_queries(**kwargs)

    filter_list = self._get_filter_list(filters)
    jobs_iter = itertools.chain(jobs['in_flight_queries'], jobs['completed_queries'])
    jobs_iter_filtered = self._n_filter(filter_list, jobs_iter)

    apps = {
      'apps': sorted([{
        'id': job['query_id'],
        'name': job['stmt'].replace('\r\n', ' ')[:60] + ('...' if len(job['stmt']) > 60 else ''),
        'status': job['state'],
        'apiStatus': self._api_status(job['state']),
        'type': job['stmt_type'],
        'user': job['effective_user'],
        'queue': job.get('resource_pool'),
        'progress': job['progress'],
        'isRunning': job['start_time'] > job['end_time'],
        'canWrite': job in jobs['in_flight_queries'],
        'duration': self._time_in_ms_groups(re.search(r"\s*(([\d.]*)([a-z]*))(([\d.]*)([a-z]*))?(([\d.]*)([a-z]*))?", job['duration'], re.MULTILINE).groups()),
        'submitted': job['start_time'],
        # Extra specific
        'rows_fetched': job['rows_fetched'],
        'waiting': job['waiting'],
        'waiting_time': job['waiting_time']
      } for job in jobs_iter_filtered], key=lambda job: job.get('submitted'), reverse=True),
      'total': 0
    }
    apps['total'] = len(apps['apps'])

    return apps

  def _time_in_ms_groups(self, groups):
    time = 0
    for x in range(0, len(groups), 3):
      if groups[x+1]:
        time += self._time_in_ms(groups[x+1], groups[x+2])
    return time

  def _time_in_ms(self, time, period):
    if period == 'ns':
      return float(time) / 1000
    elif period == 'ms':
      return float(time)
    elif period == 's':
      return float(time) * 1000
    elif period == 'm':
      return float(time) * 60000 #1000*60
    elif period == 'h':
      return float(time) * 3600000 #1000*60*60
    elif period == 'd':
      return float(time) * 86400000  # 1000*60*60*24
    else:
      return float(time)

  def app(self, appid):
    apps = self.apps({
      'text': 'id:' + appid
    })

    if not apps.get('apps'):
      return {
        'name': _('Unknown or expired query id %s') % appid
      }
    app = apps.get('apps')[0]
    progress_groups = re.search(r"([\d\.\,]+)%", app.get('progress'))
    app.update({
      'progress': float(progress_groups.group(1)) if progress_groups and progress_groups.group(1) else 100 if self._api_status(app.get('status')) in ['SUCCEEDED', 'FAILED'] else 1,
      'type': 'queries',
      'doc_url': "%s/query_plan?query_id=%s" % (self.api.url, appid),
      'properties': {
        'memory': '',
        'profile': '',
        'plan': '',
        'backends': '',
        'finstances': '',
        'metrics': ''
      }
    })

    return app

  def action(self, appid, action):
    message = {'message': '', 'status': 0}

    if action.get('action') == 'kill':

      for _id in appid:
        result = self.api.kill(_id)
        if result.get('error'):
          message['message'] = result.get('error')
          message['status'] = -1
        elif result.get('contents') and message.get('status') != -1:
          message['message'] = result.get('contents')

    return message;


  def logs(self, appid, app_type, log_name=None, is_embeddable=False):
    return {'logs': ''}

  def profile(self, appid, app_type, app_property, app_filters):
    if app_property == 'memory':
      return self._memory(appid, app_type, app_property, app_filters)
    elif app_property == 'profile':
      return self._query_profile(appid)
    elif app_property == 'backends':
      return self._query_backends(appid)
    elif app_property == 'finstances':
      return self._query_finstances(appid)
    else:
      return self._query(appid)

  def profile_encoded(self, appid):
    return self.api.get_query_profile_encoded(query_id=appid)

  def _memory(self, appid, app_type, app_property, app_filters):
    return self.api.get_query_memory(query_id=appid);

  def _metrics(self, appid):
    query_profile = self.api.get_query_profile_encoded(appid)
    profile = analyzer.analyze(analyzer.parse_data(query_profile))
    ANALYZER.pre_process(profile)
    metrics = analyzer.metrics(profile)

    if ENABLE_QUERY_ANALYSIS.get():
      result = ANALYZER.run(profile)
      if result and result[0]:
        for factor in result[0]['result']:
          if factor['reason'] and factor['result_id'] and metrics['nodes'].get(factor['result_id']):
            metrics['nodes'][factor['result_id']]['health'] = factor['reason']
    return metrics

  def _query(self, appid):
    query = self.api.get_query(query_id=appid)
    query['summary'] = query.get('summary').strip() if query.get('summary') else ''
    query['plan'] = query.get('plan').strip() if query.get('plan') else ''
    try:
      query['metrics'] = self._metrics(appid)
    except Exception, e:
      query['metrics'] = {'nodes' : {}}
      LOG.exception('Could not parse profile: %s' % e)

    if query.get('plan_json'):
      def get_exchange_icon (o):
        if re.search(r'broadcast', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-broadcast' }
        elif re.search(r'hash', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-random' }
        else:
          return { 'svg': 'hi-exchange' }
      def get_sigma_icon (o):
        if re.search(r'streaming', o['label_detail'], re.IGNORECASE):
          return { 'svg': 'hi-sigma' }
        else:
          return { 'svg': 'hi-sigma' }
      mapping = {
        'TOP-N': { 'type': 'TOPN', 'icon': { 'svg': 'hi-filter' } },
        'SORT': { 'type': 'SORT', 'icon': { 'svg': 'hi-sort' } },
        'MERGING-EXCHANGE': {'type': 'EXCHANGE', 'icon': { 'fn': get_exchange_icon } },
        'EXCHANGE': { 'type': 'EXCHANGE', 'icon': { 'fn': get_exchange_icon } },
        'SCAN HDFS': { 'type': 'SCAN_HDFS', 'icon': { 'svg': 'hi-copy' } },
{'fraction_non_alphanumeric': 0.15673881121113453, 'fraction_numerical': 0.010981293494222052, 'mean_word_length': 3.324682495858642, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '1078210', 'n_tokens_mistral': 5262, 'n_tokens_neox': 5077, 'n_words': 1251}